|| || || || || || || ||

---

ChannelServer.h

Go to the documentation of this file.

#ifndef Impala_Util_ChannelServer_h
#define Impala_Util_ChannelServer_h

#include <strstream>
#include <map>

#include "Basis/ILog.h"
#include "Basis/Timer.h"
#include "Util/Channel.h"
#include "Util/ChannelAuthorization.h"

namespace Impala
{
namespace Util
{

/**********************************************************************
 Instances of this class support communication with clients, based on 
 channels that are based on sockets.
***********************************************************************/

class ChannelServer : public ChannelListener
{

public: 

    ChannelServer(int port, int nrPorts, bool doDynaPorts, CString passwordFile)
        : mFirstPort(port), mNrOfPorts(nrPorts), mDoDynaPorts(doDynaPorts),
          mPasswordFile(passwordFile)
    {
        Init();
    }

    virtual
    ~ChannelServer()
    {
        if (mChannel)
            delete mChannel;
    }

    virtual void 
    HandleInfo(CString msg)
    {
        ILOG_INFO(msg);
    }

    virtual void 
    HandleError(CString msg)
    {
        ILOG_ERROR(msg);
    }

    virtual void 
    HandleIdle()
    {
        UpdateStatusBuf();
    }

    virtual int 
    HandleRequest(char* buf, int len, int bufSize)
    {
        buf[len] = '\0';
        String bufStr(buf, len - 1); // exclude '\0'
        String bufStrPart = (len > 20) ? bufStr.substr(0, 20) : bufStr;
        ILOG_DEBUG("Request from " << ConnectionDescr()
                   << " (" << len << " bytes) : [" << bufStrPart << "]");

        CString reqConn = mChannel->GetCurRequestConnection();
        int reqPort = mChannel->GetCurRequestPort();
        int responseLen = AcceptRequest(buf, len, bufSize, reqConn, reqPort);
        if (responseLen < 0)
        {
            String bufStr(buf, len - 1); // exclude '\0'
            String bufStrPart = (len > 20) ? bufStr.substr(0, 20) : bufStr;
            ILOG_WARN("Unknown request from " << ConnectionDescr() << " : [" <<
                      bufStrPart << "]");
            sprintf(buf, "ERROR : Unknown request\0");
            responseLen = strlen(buf) + 1;
        }

        // SK: guard against buffer overflow
        return Min(responseLen, Channel::DATA_BUFFER_SIZE);
    }

    //virtual void 
    //HandleConnect(CString conn, int port)
    //{
    //    ...
    //}

    virtual void 
    HandleDisconnect(int port)
    {
        if (mDoDynaPorts && port >= mFirstDynaPort)
        {
            if (mPortAssignment[port] != "")
            {
                AcceptDisconnect(port);
                mPortAssignment[port] = "";
                mPortAllowWrite[port] = false;
                ILOG_INFO("Port " << port << " is free and available again");
            }
        }
    }


    virtual void
    Start(bool doIdle)
    {
        mChannel = new Channel(mFirstPort, mNrOfPorts, this);
        mRunning = true;
        mTimer.Start();
        mChannel->Serve(doIdle);
        //Stop(); // SK: yet to be elaborated
    }

    // SK: yet to be elaborated
    //virtual void
    //Stop()
    //{
    //    mTimer.Stop();
    //    //mChannel...Stop() ??
    //    mRunning = false;
    //}

    int 
    NrOfPorts()
    {
        return mNrOfPorts;
    }

    int 
    FirstPort()
    {
        return mFirstPort;
    }

    bool 
    UsesDynamicPortAssignment()
    {
        return mDoDynaPorts;
    }

protected:

    virtual int
    AcceptRequest(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();

        if (strncmp(buf, "ShutDown", 8) == 0)
        {
            ILOG_WARN("Received shut down request from " << curCon <<
                      "; shutting down now");
            exit(-99);
        }

        else if (strncmp(buf, "ping", 4) == 0)
        {
            ILOG_DEBUG("Received ping request from " << curCon);
            sprintf(buf, "OK\0");
        }

        else if ((port > mFirstPort) && mDoDynaPorts &&
                 (conn != mPortAssignment[port]))
        {
            ILOG_WARN("Illegal request from " << conn << " at port " << port <<
                      "; port assigned to " << mPortAssignment[port]);
            sprintf(buf, "ERROR\0");
        }

        else if (strncmp(buf, "PortMode", 8) == 0)
            AcceptRequestPortMode(buf, len, bufSize, conn, port);

        else if (strncmp(buf, "PortAssignment", 14) == 0)
            AcceptRequestPortAssignment(buf, len, bufSize, conn, port);

        else
            return -1;

        return strlen(buf) + 1;
    }

    virtual void
    AcceptRequestPortMode(char* buf, int len, int bufSize, CString conn,
                          int port)
    {
        String curCon = ConnectionDescr();
        ILOG_DEBUG("Receiving port mode request from " << curCon);
        int index = 14;
        String readPass = ReadQuotedString(buf, index, len);
        if (! (mAuthorization.Valid() && mAuthorization.AllowRead(readPass)))
        {
            ILOG_WARN("Authorization failed");
            sprintf(buf, "Permission denied\0");
            return;
        }

        if (mDoDynaPorts)
        {
            sprintf(buf, "Dynamic\0");
            if (port != mFirstPort)
            {
                ILOG_WARN("Port mode request from " << curCon <<
                          " should have been targeted at port " << mFirstPort);
            }
        }
        else
        {
            sprintf(buf, "Static\0", port);
            index += 7;
            String writePass = ReadQuotedString(buf, index, len);
            mStaticPortAllowWrite = mAuthorization.AllowWrite(writePass);
        }
    }

    virtual void
    AcceptRequestPortAssignment(char* buf, int len, int bufSize, CString conn,
                                int port)
    {
        String curCon = ConnectionDescr();
        ILOG_DEBUG("Receiving port assignment request from " << curCon);
        if (mDoDynaPorts)
        {
            if (port == mFirstPort)
            {
                int index = 21;
                String writePass = ReadQuotedString(buf, index, len);

                int dynaPort = 0;
                for (dynaPort = mFirstDynaPort; dynaPort <= mLastDynaPort;
                     dynaPort++)
                {
                    if (mPortAssignment[dynaPort].empty())
                    {
                        ILOG_INFO("Assigning port " << dynaPort << " to " <<
                                  curCon);
                        mPortAssignment[dynaPort] = conn;
                        mPortAllowWrite[dynaPort] =
                            mAuthorization.AllowWrite(writePass);
                        sprintf(buf, "PortAssigned=%i\0", dynaPort);
                        break;
                    }
                }
                if (dynaPort > mLastDynaPort)
                {
                    ILOG_WARN("Cannot assign port: all ports are in use");
                    sprintf(buf, "NoPortAvailable\0");
                }
            }
            else
            {
                ILOG_WARN("Invalid port assignment request from " << curCon <<
                          "; should request at port " << mFirstPort);
                sprintf(buf, "ERROR\0");
            }
        }
        else
        {
            ILOG_WARN("No dynamic port assignment; no need for " << curCon <<
                      " to request port assignment");
            sprintf(buf, "PortAssigned=%i\0", port);
        }
    }

    virtual void 
    AcceptDisconnect(int port)
    {
    }

    virtual String 
    ConnectionDescr()
    {
        CString reqConn = mChannel->GetCurRequestConnection();
        int reqPort = mChannel->GetCurRequestPort();
        return reqConn + " at port " + MakeString(reqPort);
    }

    virtual void 
    UpdateStatusBuf()
    {
        double timeVal = SplitTime();
        std::ostrstream statusStream;
        statusStream << "timeVal = " << timeVal << std::ends;
        strcpy(mStatusBuf, statusStream.str());
        //SetStatusBuffer(statusStream.str());
        statusStream.freeze(false);
    }

    double 
    SplitTime()
    {
        return mTimer.SplitTime();
    }

    //void SetStatusBuffer(const char* msg)
    //{
    //    strcpy(mStatusBuf, msg);
    //}

    char* 
    GetStatusBuffer()
    {
        return mStatusBuf;
    }

    String
    ReadQuotedString(const char* buf, int& index, int len)
    {
        while (index < len && buf[index] != '"')
            index++;
        if (index < len)
        {
            int startIndex = ++index; // read past start quote
            while (index < len && buf[index] != '"')
                index++;
            const int quotedStringLen = index - startIndex;
            if (index < len)
                index++; // read past end quote
            return String(buf + startIndex, quotedStringLen);
        }
        return "";
    }

    bool
    AllowWrite(int port)
    {
        if (mDoDynaPorts)
            return mPortAllowWrite[port];
        return mStaticPortAllowWrite;
    }

private:

    void 
    Init()
    {
        mChannel = 0;
        mRunning = false;
        mTimer = 1;

        // need at least 2 ports for dynamic port assignment
        mDoDynaPorts = mDoDynaPorts && (mNrOfPorts > 1);
        if (mDoDynaPorts)
        {
            mFirstDynaPort = mFirstPort + 1;
            int nrOfDynaPorts = mNrOfPorts - 1;
            mLastDynaPort = mFirstPort + nrOfDynaPorts;
            ILOG_INFO("Using dynamic port assignment");
        }
        else
        {
            mFirstDynaPort = -1;
            mLastDynaPort = -1;
            ILOG_INFO("Not using dynamic port assignment");
        }

        mAuthorization = ChannelAuthorization(mPasswordFile);
    }

    int      mFirstPort;
    int      mNrOfPorts;
    bool     mDoDynaPorts;
    String   mPasswordFile;
    int      mFirstDynaPort;
    int      mLastDynaPort;
    //bool     mDynaPorts;
    Channel* mChannel;
    bool     mRunning;
    Timer    mTimer;
    char     mStatusBuf[256];
    std::map<int, String> mPortAssignment;
    std::map<int, bool>   mPortAllowWrite;
    bool                  mStaticPortAllowWrite;
    ChannelAuthorization  mAuthorization;

    ILOG_VAR_DEC;

}; // class

ILOG_VAR_INIT(ChannelServer, Impala.Util);

} // namespace 
} // namespace 

#endif

---