---

MsgTransport.cpp

Go to the documentation of this file.

/*
 * MsgTransport.cpp
 *
 * Copyright 2003, MobileSpear Inc. (www.mobilespear.com). All rights reserved.
 * Copyright 2003, David Resnick. All rights reserved.
 *
 * See the file doc\license.txt for the terms of usage and distribution.
 */

#include <bogotel/Portability.h>
#include <bogotel/MsgTransport.h>
#include <bogotel/BgtRt.h>
#include <bogotel/BgtErrors.h>
#include <bogotel/util.h>

#include <boost/thread/xtime.hpp>
#include <log4cpp/NDC.hh>

namespace bogotel {

    #define MAXBUFLEN 500

    // CMsgTransport

    CMsgTransport::CMsgTransport(CBgtRt *pBgtRt) : 
    m_usMyPort(0), 
    m_usOppPort(0),
    m_pBgtRt(pBgtRt),
    m_bListenerStarted(false),
    m_bStopTalker(false),
    m_bStopListener(false)
    {

    }

    CMsgTransport::~CMsgTransport()
    {
        /*
        m_bStopListener = true;
        m_bStopTalker = true;

        m_pthrdListener->join();
        delete m_pthrdListener;

        m_pthrdTalker->join();
        delete m_pthrdTalker;
        */

        /*
        shutdown(m_fdListener, SD_BOTH);
        shutdown(m_fdTalker, SD_BOTH);
        */
    }

    int CMsgTransport::init()
    {
        char method[] = "CMsgTransport::init()";

        std::string strInstanceNum = g_util->getInstanceNum();

        if ((m_usMyPort = (u_short)g_util->m_propIni.getProperty("MyPort" + strInstanceNum, (unsigned long)0)) == 0) {
            g_util->log(3, -1, "%s: No MyPort in ini file", method);
            return resultERROR;
        }

        if ((m_usOppPort = (u_short)g_util->m_propIni.getProperty("OppPort" + strInstanceNum, (unsigned long)0)) == 0) {
            g_util->log(3, -1, "%s: No OppAddress in ini file", method);
            return resultERROR;
        }

        m_strOppIp = g_util->m_propIni.getProperty("OppAddress" + strInstanceNum, "127.0.0.1");
        if (m_strOppIp == "") {
            g_util->log(3, -1, "%s: No OppAddress in ini file", method);
            return resultERROR;
        }

        {
            boost::mutex::scoped_lock lock(m_mtxStartup);
        
            // start the listener on a separate thread
            m_pthrdListener = new boost::thread(thread_adapter(&CMsgTransport::do_thread_listen, this));

            boost::xtime xt;
            boost::xtime_get(&xt, boost::TIME_UTC);
            xt.sec += 25;
            if (! m_condStartup.timed_wait(lock, xt, cond_predicate(m_bListenerStarted, true))) {
                // this is taking too long...
                g_util->log(3, -1, "%s: Timed out waiting for listener thread.", method);
                return resultERROR;
            }
        }

        // start the talker on a separate thread
        m_pthrdListener = new boost::thread(thread_adapter(&CMsgTransport::do_thread_talk, this));

        g_util->log(9, -1, "CMsgTransport::init() completed.");
        return resultSUCCESS;
    }

    void CMsgTransport::do_thread_listen(void* param)
    {
        log4cpp::NDC::push("ListenThread");
        try {
            static_cast<CMsgTransport*>(param)->runListener();
        } catch(...) {
            g_util->log(1, -1, "CMsgTransport::do_thread_listen: Exception");
        }
    }

    void CMsgTransport::do_thread_talk(void* param)
    {
        log4cpp::NDC::push("TalkThread");
        try {
            static_cast<CMsgTransport*>(param)->runTalker();
        } catch(...) {
            g_util->log(1, -1, "CMsgTransport::do_thread_talk: Exception");
        }
    }

    void CMsgTransport::runTalker()
    {
        int rc;
        int numbytes;

        if ((rc = openTalkerSocket()) != resultSUCCESS) {
            g_util->log(3, -1, "CMsgTransport::talkerRun(): openTalkerSocket() failed. rc = %d", rc);
            return;
        }

        while (!m_bStopTalker) {
            {
                boost::mutex::scoped_lock lock(m_mtxAddedToQueue);
                ContainerNotEmpty<DEQUE_STRING> notEmpty(m_deqMsg);

                // wait until a new message is posted
                m_condEvent.wait(lock, notEmpty);
            }

            while (true) {
                std::string strMsg;

                // get an item from the queue (if any)
                {
                    boost::mutex::scoped_lock lock(m_mtxDeque);
                    if (! m_deqMsg.empty()) {
                        strMsg = m_deqMsg.front();
                        m_deqMsg.pop_front();
                    } else {
                        // nothing was in the queue, wait for another message
                        break;
                    }
                }

                if ((numbytes = sendto(m_fdTalker,
                                       strMsg.c_str(),
                                       strlen(strMsg.c_str()) + 1,
                                       0,
                                       (struct sockaddr *)&m_addrTalker,
                                       sizeof(struct sockaddr))) == -1) {
                    g_util->log(3, -1, "CMsgTransport::talkerRun(): sendto() failed. "
                        "::WSAGetLastError() is %d", ::WSAGetLastError());
                    return;
                }
            }
        }
    }

    void CMsgTransport::runListener()
    {
       int rc;
       int addr_len, numbytes;
       char buf[MAXBUFLEN];

        // open socket for listening
        if ((rc = openListenerSocket(&m_addrListener, &m_fdListener, m_usMyPort)) != resultSUCCESS) {
            g_util->log(3, -1, "CMsgTransport::listenerRun(): openListenerSocket() failed. rc = %d", rc);
            return;
        }

        // inform main thread that we've finished initializing successfully
        m_bListenerStarted = true;
        m_condStartup.notify_one();

        CMsg *pMsg = new CMsg();

        // listen for packets on socket
        while (! m_bStopListener) {
            addr_len = sizeof(struct sockaddr);
            if ((numbytes = recv(m_fdListener, buf, MAXBUFLEN, 0)) == -1) {
                g_util->log(3, -1, "CMsgTransport::listenerRun(): recvfrom() failed. "
                    "::WSAGetLastError() is %d", ::WSAGetLastError());
                continue;
            }

            if ((rc = pMsg->init(buf)) != resultSUCCESS) {
                g_util->log(3, -1, "CMsgTransport::listenerRun(): CMsg::init() failed. rc is %d", rc);
            } else {
                m_pBgtRt->incomingMsg(pMsg);
            }
        }
    }

    int CMsgTransport::openListenerSocket(struct sockaddr_in *pAddr,
                                                int *pFd,
                                                u_short usPort)
    {
        char method[] = "CMsgTransport::openListenerSocket()";
        if (((*pFd) = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
            g_util->log(3, -1, "%s: socket() failed. ::WSAGetLastError() is %d", 
                method, ::WSAGetLastError());
            return resultSOCKET_ERROR;
        }
        pAddr->sin_family = AF_INET;         // host byte order 
        pAddr->sin_addr.s_addr = INADDR_ANY; // auto-fill with my IP 
        memset(&(pAddr->sin_zero), 0, 8);    // zero the rest of the struct 
        pAddr->sin_port = htons(usPort);     // short, network byte order 

        if (bind((*pFd), (struct sockaddr *)pAddr, sizeof(struct sockaddr)) != 0) {
            // some error, check if the port is in use
            int iErr = ::WSAGetLastError();
            if (iErr == WSAEADDRINUSE) {
                g_util->log(5, -1, "%s: bind() failed. Socket in use.", method);
                return resultSOCKET_IN_USE;
            }
            g_util->log(3, -1, "%s: bind() failed. ::WSAGetLastError() is %d", 
                method, ::WSAGetLastError());
            return resultSOCKET_ERROR;
        }
        g_util->log(9, -1, "%s completed.", method);
        return resultSUCCESS;
    }

    int CMsgTransport::openTalkerSocket()
    {
        struct hostent *pheTalker = NULL;

        // prepare the address to be talked to
        if ((pheTalker = gethostbyname(m_strOppIp.c_str())) == NULL) {  /* get the host info */
            g_util->log(3, -1, "CMsgTransport::openTalkerSocket(): gethostbyname() failed. "
                "::WSAGetLastError() is %d", ::WSAGetLastError());
            return resultSOCKET_ERROR;
        }
        m_addrTalker.sin_family = AF_INET;      /* host byte order */
        m_addrTalker.sin_port = htons(m_usOppPort);  /* short, network byte order */
        m_addrTalker.sin_addr = *((struct in_addr *)pheTalker->h_addr);
        memset(&(m_addrTalker.sin_zero), 0, 8);     /* zero the rest of the struct */

        if ((m_fdTalker = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
            g_util->log(3, -1, "CMsgTransport::openTalkerSocket(): socket() failed. "
                "::WSAGetLastError() is %d", ::WSAGetLastError());
            return resultSOCKET_ERROR;
        }

        g_util->log(9, -1, "CMsgTransport::openTalkerSocket() completed.");
        return resultSUCCESS;
    }

    int CMsgTransport::sendMsg(CMsg *pMsg)
    {
        char szMsg[MAXBUFLEN];

        if (pMsg->toString(szMsg, sizeof(szMsg)) == NULL){
            g_util->log(3, -1, "CMsgTransport::sendMsg(): CMsg::toString() failed. ");
            return resultFUNC_BAD_PARAMETER;
        }

        // put the string in the queue
        {
            boost::mutex::scoped_lock lock(m_mtxDeque);
            m_deqMsg.push_back(szMsg);
        }

        // wake the talker thread up so that it sends the message
        m_condEvent.notify_one();

        g_util->log(9, -1, "CMsgTransport::sendMsg() completed.");

        return resultSUCCESS;
    }

    // CMsg

    CMsg::CMsg()
    {
        clear();
    }

    CMsg::~CMsg()
    {

    }

    void CMsg::clear()
    {
        m_type = MT_INVALID;
        m_class = MC_INVALID;
        m_id = -1;
        m_crn = INVALID_CRN;
        if (! m_mapParam.empty()) {
            m_mapParam.clear();
        }
    }

    void CMsg::init(MessageType type, MessageClass cls, long lId)
    {
        clear();
        m_class = cls;
        m_type = type;
        m_id = lId;
    }

    void CMsg::addParam(enum ParmType pt, std::string strParm)
    {
        m_mapParam.insert(MAP_LONG2STR::value_type(pt, strParm));
    }

    void CMsg::addParam(enum ParmType pt, char *szParm)
    {
        m_mapParam.insert(MAP_LONG2STR::value_type(pt, szParm));
    }

    std::string CMsg::getParam(enum ParmType pt)
    {
        MAP_LONG2STR::iterator itL2S;
        itL2S = m_mapParam.find(pt);
        if (itL2S == m_mapParam.end()) {
            return "";
        }
        return itL2S->second;
    }

    // serializes the message to a null terminated string
    char *CMsg::toString(char *szDmp, size_t iDmpLen)
    {
        char szTmp[MAXBUFLEN];
        int rc;
        rc = _snprintf(szDmp, iDmpLen, "Cls:%d Crn:%d Id:%d Type:%d", m_class, m_crn, m_id, m_type);
        if (rc < 0) {
            return NULL;
        }
        MAP_LONG2STR::iterator it;
        it = m_mapParam.begin();
        while (it != m_mapParam.end()) {
            long p1 = it->first;
            std::string p2 = it->second;
            if (p2 == "") {
                g_util->log(3, -1, "CMsg::toString(): parm data empty. parm type is %d", p1);
                return NULL;
            }
            rc = _snprintf(szTmp, sizeof(szTmp), " ;PT:%d PC:%s", p1, p2.c_str());
            if (rc < 0) {
                g_util->log(3, -1, "CMsg::toString(): _snprintf failed. (probably should increase size of szTmp)");
                return NULL;
            }
            if (strlen(szTmp) + strlen(szDmp) + 1 > iDmpLen) {
                g_util->log(3, -1, "CMsg::toString(): szDmp is too small "
                    "(must be inceased to at least %d)", strlen(szTmp) + strlen(szDmp) + 1);
                return NULL;
            }
            strcat(szDmp, szTmp);
            it++;
        }
        return szDmp;
    }

    int CMsg::init(char *szMsg)
    {
        char _func[] = "CMsg::init";
        // first clear the class in case it's being reused
        if (m_class != MC_INVALID) {
            clear();
        }

        int rc = sscanf(szMsg, "Cls:%d Crn:%d Id:%d Type:%d", &m_class, &m_crn, &m_id, &m_type);
        if (rc != 4) {
            g_util->log(3, -1, "%s: failed to retrieve 4 vars from msg \"%s\"", _func, szMsg);
            return resultFUNC_BAD_PARAMETER;
        }

        // check for parameters
        char *szTok = strtok(szMsg, ";");
        if (szTok == NULL) {
            // there are no parameters, return
            return resultSUCCESS;
        }

        enum ParmType pt;
        char szTemp[MAXBUFLEN];
        // get the next token and discard the initial token (it was already handled above)
        while ((szTok = strtok(NULL, ";")) != NULL) {
            // get the parm type and data from the token
            rc = sscanf(szTok, "PT:%d PC:%s", &pt, szTemp);
            if (rc != 2) {
                g_util->log(3, -1, "%s: failed to retrieve parm type/data from token \"%s\"", _func, szTok);
                return resultFUNC_BAD_PARAMETER;
            }
            addParam(pt, szTemp);
        }
        return resultSUCCESS;
    }

    void CMsg::getMsgTypeString(char *szType, int iMaxLen)
    {
        switch(m_type) {
        case MT_INVALID:           strncpy(szType,"INVALID", iMaxLen);     break;
        case MT_MAKE_CALL:         strncpy(szType,"MAKE_CALL", iMaxLen);   break;
        case MT_ACCEPT_CALL:       strncpy(szType,"ACCEPT_CALL", iMaxLen); break;
        case MT_ANSWER_CALL:       strncpy(szType,"ANSWER_CALL", iMaxLen); break;
        case MT_DROPCALL:          strncpy(szType,"DROP_CALL", iMaxLen);   break;
        case MT_RELEASE_CALL:      strncpy(szType,"RELEASE_CALL", iMaxLen);break;
        case MT_PLAY_WAV:          strncpy(szType,"PLAY_WAV", iMaxLen);    break;
        case MT_PLAY_WAV_FINISHED: strncpy(szType,"PLAY_WAV_FINISHED", iMaxLen);break;
        case MT_PLAY_DTMF:         strncpy(szType,"PLAY_DTMF", iMaxLen);    break;
        case MT_PLAY_DTMF_FINISHED:strncpy(szType,"PLAY_DTMF_FINISHED", iMaxLen);break;
        default:                   strncpy(szType,"Unknown", iMaxLen);     break;
        }
    }

}

---