siptcpconnection.cxx

Vovida 社区开源的 SIP 协议源码

        }

        // this is a request
#ifdef USE_PERSISTENT_TCP
////        conn = createOrGetPersistentConnection(sipCommand);
        conn = createOrGetPersistentConnection(*(sipMsg->msg.netAddr));
#else
        if(sipCommand->nextHopIsAProxy())
        {
	  
////            conn = createOrGetPersistentConnection(sipCommand);
            conn = createOrGetPersistentConnection(*(sipMsg->msg.netAddr));
        }
        else
        {
             conn = createRequestTransaction(sipMsg);
        }
#endif
    }
    else
    {
        // this is an error
        cpLog(LOG_ERR, "API violation: attempt to send null message");
        return;
    }
    // at this point, you need to transmit
    if (conn != 0)
    {
        sipMsg->msg.out =  sipMsg->msg.in->encode();
        string dataString = sipMsg->msg.out.getData(lo);
        cpLog(LOG_INFO, "sending TCP SIP msg:\n\n-> %s\n\n%s",
              conn->getDescription().c_str(), dataString.c_str());
        cpLog(LOG_DEBUG_STACK, "TCP fd: %d", conn->getConnId());

        conn->writeData(dataString);

#ifdef USE_PERSISTENT_TCP
        Sptr<StatusMsg> statusMsg;
        statusMsg.dynamicCast(sipMsg->msg.in);
        if((statusMsg != 0) && (statusMsg->getStatusLine().getStatusCode() >= 200))
        {
            //Transaction is over, remove the entry from the ID map
            SipTransactionId id(*statusMsg);
            tcpConnInfo.delIdMapEntry(id);
        }
#else
        if((type == SIP_ACK) && !(sipMsg->msg.in->nextHopIsAProxy()))
        {
            LockHelper l(tcpConnInfo.cleanupMutex);
            tcpConnInfo.myCleanupList.push_back(conn->getConnId());
        }
#endif
    }
}


string
get_next_line(const string& str, string::size_type* old_pos)
{
    string myStr;

    string::size_type pos = str.find('\n', *old_pos);
    if (pos != string::npos)
    {
        myStr = str.substr(*old_pos, pos - (*old_pos));
        *old_pos = pos + 1;
    }
    else
    {
        myStr = str.substr(*old_pos, str.length() - *old_pos);
        *old_pos = pos;
    }
    return myStr;
}


int
getContentLength(const string& str)
{
    LocalScopeAllocator lo;
    // given that this string is a content-length then do something
    static string contentLengthStr = str2lower(CONTENT_LENGTH.getData(lo));
    static string contentLengthStr2 = str2lower(CONTENT_LENGTH_SHORT.getData(lo));
   
    int pos = 0; 
    unsigned len = str.length();
    if(len > contentLengthStr.length())
         len = contentLengthStr.length();
    string x = str2lower(str.substr(0,len));

    if (x.find(contentLengthStr) == 0) 
    {
        // this is right -- scan
        pos = contentLengthStr.length();
    }
    else if (x.find(contentLengthStr2) == 0) 
    {
        pos = contentLengthStr2.length();
    }
    else
    {
        // not the content-length
        return -1;
    }
    string::size_type newPos = str.find_first_not_of("0123456789", pos + 1);
    if (newPos == string::npos)
    {
        newPos = str.length();
    }
    string contentLen = str.substr(pos + 1, newPos - (pos + 1));
    return (atoi(contentLen.c_str()));
}


bool
isBlankLine(const string& str)
{
    // a line which has only whitespace is NOT a blank line for the
    // purposes of separating the headers from the body in a SIP
    // message, only a line which has ONLY the CRLF.  we also accept
    // lines w/ LF only for robustness.

    if((str == "\r") || (str == ""))
    {
        return true;
    }
    else
    {
        return false;
    }

#if 0
    if (str.find_first_not_of(" \t\r\n") == string::npos)
    {
        return true;
    }
    else
    {
        return false;
    }
#endif
}


int 
Vocal::isFullMsg(const string& str)
{
    string::size_type pos = 0;

    int contentLen = -1;
    bool foundBlank = false;
    bool foundNonBlank = false;
    LocalScopeAllocator lo;
    pos = str.find(CONTENT_LENGTH.getData(lo));
    if(pos == string::npos)
    {
        pos = str.find(CONTENT_LENGTH_SHORT.getData(lo));
        if(pos == string::npos) pos = 0;
    }

    if(pos != 0)
    {
        foundNonBlank = true;
    }

    while (!foundBlank && (pos != string::npos))
    {
        string myLine = get_next_line(str, &pos);

        if(pos != string::npos)
        {
            if (contentLen == -1)
            {
                // check against content-length
                contentLen = getContentLength(myLine);
            }
            if(isBlankLine(myLine))
            {
                if(foundNonBlank)
                {
                    foundBlank = true;
                    break;
                }
            }
            else
            {
                foundNonBlank = true;
            }
        }
    } // While

    if (pos == string::npos)
    {
        pos = str.length();
    }

    if (foundBlank)
    {
        // then make sure that the remaining character count is the same
        int remainingBytes = str.length() - pos;

        // if there have been no blank lines, ignore these lines
        if (contentLen == -1 && foundNonBlank)
        {
            contentLen = 0;
        }
        if ((remainingBytes >= contentLen) && (str.length() > 0))
        {
            return (pos + contentLen);
        }
    }
    return -1;
}


void
SipTcpConnection_impl_::processMsgsIfReady(int fd)
{
    Data data;
    string ipName;

    // read off some data, possibly break
    Sptr<NTcpStuff> connInfo = tcpConnInfo.getConnInfo(fd);
    if(connInfo == 0)
    {
        // Too late , connection is closed already
        return;
    }
    LockHelper l(connInfo->myMutex);
    LocalScopeAllocator lo;
    Data& myBufD = connInfo->tcpBuf;
    ipName = connInfo->senderIp.getData(lo);
    string myBuf = myBufD.getData(lo);
    int bufLen = isFullMsg(myBuf);
    cpLog(LOG_DEBUG_STACK, "Raw Msg read: [%s]" , myBuf.c_str());
    while (bufLen != -1)
    {
        string msgBuf = myBuf.substr(0, bufLen);
        string remainingBuf =
            myBuf.substr(bufLen, myBuf.length() - bufLen);
        myBufD = Data(remainingBuf);

        data = Data(msgBuf);

        SipMsgContainer* sipMsg = new SipMsgContainer;



	sipMsg->msg.out = data;
        sipMsg->msg.in = SipMsg::decode( data );

        if (sipMsg->msg.in != 0)
        {
            //store the receivedIPName.
            cpLog(LOG_DEBUG_STACK, "*** getSenderIP = %s***", ipName.c_str());
            sipMsg->msg.in->setReceivedIPName( ipName );
            cpLog(LOG_INFO, "Received UDP Message :\n\n<- HOST[%s]\n\n%s",ipName.c_str(),  data.logData());

            Sptr<StatusMsg> statusMsg;
            statusMsg.dynamicCast(sipMsg->msg.in);
            if ((sipMsg->msg.in->getType() != SIP_STATUS) &&
                (sipMsg->msg.in->getType() != SIP_ACK))
            {
                // save in the map
                tcpConnInfo.setStatusMsgConn(sipMsg->msg.in, fd);
                if(sipMsg->msg.in->getType() == SIP_REGISTER)
                {
                   //If request is a REGISTER message, map the contact
                   //host port with TCP socket, so that future
                   //requests to the contact would follow the same
                   //TCP connection
                   Data rData = connInfo->tcpConnection->getDescription();
                   cpLog( LOG_DEBUG_STACK, "Received from %s", rData.logData() );
                   int numContacts = sipMsg->msg.in->getNumContact();
                   if( numContacts )
                   {
                       const SipContact& contact = sipMsg->msg.in->getContact( numContacts-1 );
                       Sptr<SipUrl> sUrl;
                       sUrl.dynamicCast( contact.getUrl() );
                       if( sUrl != 0 )
                       {
                           Data tmp = sUrl->getHost();
                           tmp += ":";
                           tmp += sUrl->getPort();
                           string cAddr(tmp.getData(lo));
                           cpLog( LOG_DEBUG_STACK, "fd %d for address %s", fd, cAddr.c_str() );
                           myDestinationMap[ cAddr ] = new Connection( fd );
                       }
                   }
                }
            }
            else if((statusMsg != 0) && 
                    statusMsg->getStatusLine().getStatusCode() >= 200)
            {
#ifndef USE_PERSISTENT_TCP
                string iName = connInfo->tcpConnection->getDescription();
                if(myDestinationMap.count(iName) == 0)
                {
                    cpLog(LOG_DEBUG_STACK, 
                          "Deleting non-persistent connection %d", fd);
                    //Non-persistent connection, close
                    tcpConnInfo.delConn(fd);
                }
                //Do the cleanup also
                tcpConnInfo.doCleanup();
#endif
             }
           
            
            recFifo->add(sipMsg);
        }
	else
	  {
	    delete sipMsg;
	  }
        myBuf = myBufD.getData(lo);
        bufLen = isFullMsg(myBuf);
    } // while
}


void
SipTcpConnection_impl_::readOnFdSet(fd_set* fdSet, TcpServerSocket* tcpStack)
{
    tcpConnInfo.mutex.lock();
    int tcpreadfd = -1;
    list<int> fdList;

    map < TcpFd, Sptr < NTcpStuff > > ::iterator i = tcpConnInfo.myMap.begin();
    while (i != tcpConnInfo.myMap.end())
    {
        int mapfd = i->first;

        if (FD_ISSET(mapfd, fdSet))
        {
            FD_CLR(mapfd, fdSet);
            fdList.push_back(mapfd);
        }
        ++i;
    }
    tcpConnInfo.mutex.unlock();
    for(list<int>::iterator itr = fdList.begin(); itr != fdList.end();
             itr++)
    {
        int mapfd = (*itr);
        cpLog(LOG_DEBUG_STACK, "reading tcp data on %d", mapfd);
        tcpreadfd =  tcpConnInfo.tcpReadOrAccept(mapfd, tcpStack);
        if(tcpreadfd > 0) processFifo.add(tcpreadfd);
    }
}


void* 
SipTcpConnection_impl_::receiveMain()
{
    fd_set netFD;

    while (1)
    {
        FD_ZERO (&netFD);
        int maxfd = tcpConnInfo.setTCPFds( &netFD );

        struct timeval timeout;
        timeout.tv_sec = 0;
        timeout.tv_usec = 10000;
        int selret = select(maxfd + 1, &netFD, 0, 0, &timeout);

        if (selret > 0)
        {
            readOnFdSet(&netFD, &mytcpStack);
        }
	else if (selret < 0)
        {
            // do something
            if ( errno != EINTR )
            {
	        // this should be an error
                cpLog(LOG_ALERT, "select() failed, reason: %s, %d", 
                      strerror(errno), errno);
            }
        }
	else if (selret == 0)
        {
            if (shutdownNow)
            {
                return 0;
            }
        }
    }
    return 0;
}


SipTcpConnection::SipTcpConnection( Fifo <SipMsgContainer* > * fifo, int port)
{
    impl_ = new SipTcpConnection_impl_(fifo, port);
}


SipTcpConnection::~SipTcpConnection()
{
    delete impl_;
}


void
SipTcpConnection::send(SipMsgContainer* msg, const Data& host,
                             const Data& port)
{
    impl_->send(msg, host, port);
}


/* Local Variables: */
/* c-file-style: "stroustrup" */
/* indent-tabs-mode: nil */
/* c-file-offsets: ((access-label . -) (inclass . ++)) */
/* c-basic-offset: 4 */
/* End: */