tpkt.c

基于h323协议的软phone

* tpktEvent
*
* Purpose: Treats the events that may occure on a connection (Accept, Connect, Write, Read,
*      close).
*
* Input:   socket  - The listenning socket (in case of accept event)
*          event   - The type of the event
*          error   - indicates whether an error occured in the li level
*          context - The tpkt object
*
* Output:  None.
*******************************************************************************************/
void tpktEvent(
    IN RvSelectEngine*  selectEngine,
    IN RvSelectFd*      fd,
    IN RvSelectEvents   selectEvent,
    IN RvBool           error)
{
    RvH323Connection * connection = RV_GET_STRUCT(RvH323Connection, fd, fd);
    tpktInfo* tpkt = RV_GET_STRUCT(tpktInfo, connection, connection);
    tpktGlobals* globals = (tpktGlobals*) connection->context;
    RvStatus ret = RV_OK;

    RV_UNUSED_ARG(selectEngine);

    if (!fd) return;
    if (!globals) return;

    if (!emaLock((EMAElement)tpkt))
        return;

    /* Treat the events on the given socket */
    switch(selectEvent)
    {
        /* accept was done in order to open and incoming connection request */
        case RvSelectAccept:
        {
            RvSocket newSocket;
            RvAddress remoteAddress;
            int numOfLocks;

            RvLogDebug(&globals->log, (&globals->log,
                "tpktEvent(connection=%p, RvSelectAccept, error=%d, context=%p)", connection, error, globals));

             /* Notify application that there is incoming connection request */
            numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
            tpkt->send.eventHandler((HTPKT)tpkt,(RvSelectEvents)0,0,RV_FALSE,tpkt->send.context);
            emaReturnFromCallback((EMAElement)tpkt, numOfLocks);

            /*if during the event handler tpkt was not released then accept socket */
            if (!emaWasDeleted((EMAElement)tpkt))
                /* Do the accept and get the connection socket */
                ret = RvSocketAccept(&connection->socket, &newSocket, &remoteAddress);
            else
            {
                emaUnlock((EMAElement)tpkt);
                return;
            }
            if (ret != RV_OK)
            {
                int numOfLocks;
                /* call the callback with error indication */
                numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                tpkt->send.eventHandler((HTPKT)tpkt,selectEvent,0,RV_TRUE,tpkt->send.context);
                emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
            }
            else
            /* We want to leave the listenning socket active */
            if (tpkt->type==tpktMultiServer)
            {
                tpktInfo *newTpkt = (tpktInfo *)emaAdd(globals->hEma, NULL);

                /* check that a new element was allocated */
                if (newTpkt == NULL)
                {
                    int numOfLocks;
                    RvSocketDestruct(&newSocket, RV_TRUE, globals->pPortRange);
                    numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                    tpkt->send.eventHandler((HTPKT)tpkt,selectEvent,0,RV_TRUE,tpkt->send.context);
                    emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
                    break;
                }

                /* initialize the new tpktInfo element */
                memset(newTpkt, 0, sizeof(tpktInfo));
                newTpkt->close             = RV_FALSE;
                newTpkt->type              = tpktServer; /* The connection type is tpktServer for the individiual connection */
                newTpkt->event             = RvSelectClose; /* set the ability to receive close events on the connection */

                newTpkt->connection.type     = tpkt->connection.type;
                newTpkt->connection.context  = tpkt->connection.context;
                newTpkt->connection.socket   = newSocket;
                RvFdConstruct(&newTpkt->connection.fd, &newTpkt->connection.socket);

                if (!error) newTpkt->isConnected = RV_TRUE;

                /* register the event (close event) for the connection */
                RvSelectAdd(globals->pSelEngine, &newTpkt->connection.fd, newTpkt->event, tpktEvent);
                /* notify the appl that the accept was done */
                tpkt->send.eventHandler((HTPKT)newTpkt,selectEvent,0,RV_FALSE,tpkt->send.context);
            }
            else
            /* we don't want the listenning socket anymore */
            {
                int numOfLocks;

                RvSelectRemove(globals->pSelEngine, &connection->fd);
                RvFdDestruct(&connection->fd);
                RvSocketDestruct(&connection->socket, RV_FALSE, globals->pPortRange);

                if (!error) tpkt->isConnected = RV_TRUE;

                /* use the object created as the connection object */
                tpkt->connection.socket = newSocket;
                tpkt->event = RvSelectClose; /* set the ability to get close event on the connection */
                RvFdConstruct(&tpkt->connection.fd, &tpkt->connection.socket);
                RvSelectAdd(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

                /* notify that a connection was created */
                numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                tpkt->send.eventHandler((HTPKT)tpkt,selectEvent,0,RV_FALSE,tpkt->send.context);
                emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
            }
        }
        break;

        /* Treat a connect event */
        case RvSelectConnect:
        {
            int numOfLocks;

            RvLogDebug(&globals->log, (&globals->log,
                "tpktEvent(connection=%p, RvSelectConnect, error=%d, context=%p)", connection, error, globals));

            /* remove the connect event from the allowed events bitmap and add a close event ability*/
            tpkt->event = (RvSelectEvents) (tpkt->event & ~RvSelectConnect);
            tpkt->event = (RvSelectEvents) (tpkt->event | RvSelectClose);
            /* mark the connection as open */
            if (!error) tpkt->isConnected = RV_TRUE;
            /* register the events for the connection */
            RvSelectUpdate(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

            /* notify the appl that the connect was done */
            numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
            tpkt->send.eventHandler((HTPKT)tpkt,selectEvent,0,error,tpkt->send.context);
            emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
        }
        break;

        /* treat a write event */
        case RvSelectWrite:
        {
            int  sent;
            RvBool finished = RV_FALSE;

            RvLogDebug(&globals->log, (&globals->log,
                "tpktEvent(connection=%p, RvSelectWrite, error=%d, context=%p)", connection, error, globals));

            /* send the rest of the message */
            if ((sent = socketSendTcpIfConnected(tpkt, &finished)) < 0)
            {
                int numOfLocks;

                /* remove the write event from the connection */
                tpkt->event = (RvSelectEvents) (tpkt->event & ~RvSelectWrite);

                /* notify that the message was sent but not completed */
                RvSelectUpdate(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

                numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                tpkt->send.eventHandler((HTPKT)tpkt,selectEvent,0,RV_TRUE,tpkt->send.context);
                emaReturnFromCallback((EMAElement)tpkt, numOfLocks);

                break;
            }

            /* if ALL the message was sent already */
            if (finished)
            {
                int numOfLocks;

                /* remove the write event from the connection */
                tpkt->event = (RvSelectEvents) (tpkt->event & ~RvSelectWrite);
                RvSelectUpdate(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

                /* notify that the message was completely sent and give the number of bytes sent */
                numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                tpkt->send.eventHandler((HTPKT)tpkt,selectEvent,0,RV_FALSE,tpkt->send.context);
                emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
            }
        }
        break;

        /* treat a read event */
        case RvSelectRead:
        {
            RvSize_t recv = 0;
            RvBool finished = RV_FALSE;

            RvLogDebug(&globals->log, (&globals->log,
                "tpktEvent(connection=%p, RvSelectRead, error=%d, context=%p)", connection, error, globals));

            if (error)
            {
                /* If we had an error on a read event, we assume there's nothing to do with the
                   connection and we close it */
                tpkt->close = RV_TRUE;
            }

            /* if the connection is closing */
            if (tpkt->close)
            {
                RvSize_t rc = 1;
                int safetyCounter=0;
#ifndef _NUCLEUS
                /* read everything still on the connection and drop it */
                for(safetyCounter = 0; (safetyCounter < 10) && (rc > 0) && (ret == RV_OK); safetyCounter++)
                {
                    ret = RvSocketReceiveBuffer(&tpkt->connection.socket,(RvUint8*)tpktDummyBuff,sizeof(tpktDummyBuff),
                        &rc, NULL);
                }
#endif
                /* if all was read, the connection was emptied */
                if ((ret != RV_OK) || (rc < 0) || (safetyCounter == 10))
                {
                    /* close the connection */
                    tpktClose((HTPKT)tpkt);
                }
                break;
            }

            /* if there is no buffer */
            if (!tpkt->recv.hRpool)
            {
                int numOfLocks;
                /* notify that there was nothing read */
                numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                tpkt->recv.eventHandler((HTPKT)tpkt,selectEvent,0,RV_FALSE,tpkt->recv.context);
                emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
                break;
            }

            /* if we are still reading the header */
            if (tpkt->recv.headerCount < TPKT_HEADER_SIZE)
            {
                /* get the rest of the header */
                ret = socketRecvTcpDirectIfConnected(tpkt, (tpkt->recv.header+tpkt->recv.headerCount),
                    (TPKT_HEADER_SIZE-tpkt->recv.headerCount), &recv);
                if (ret != RV_OK)
                {
                    int numOfLocks;

                    /* On error, close down the read event capability and notify the error */
                    tpkt->event=(RvSelectEvents) (tpkt->event & ~RvSelectRead);
                    RvSelectUpdate(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

                    numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                    tpkt->recv.eventHandler((HTPKT)tpkt,selectEvent,0,RV_TRUE,tpkt->recv.context);
                    emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
                    break;
                }
                tpkt->recv.headerCount += recv;

                RvLogDebug(&globals->log, (&globals->log,
                    "Received header: %02X %02X %02X %02X",
                    tpkt->recv.header[0], tpkt->recv.header[1], tpkt->recv.header[2], tpkt->recv.header[3]));
            }

            /* if we have the header we can allocate the rpool message */
            if (tpkt->recv.headerCount == TPKT_HEADER_SIZE)
            {
                int tpktLen=(((int)(tpkt->recv.header[2])<<8)+tpkt->recv.header[3])-TPKT_HEADER_SIZE;

                if (tpktLen > 0)
                {
                    RvLockGet(tpkt->recv.rpoolLock);
                    rpoolRealloc(tpkt->recv.hRpool, tpkt->recv.message, tpktLen);
                    RvLockRelease(tpkt->recv.rpoolLock);
                }
            }

            /* we are reading a new message body */
            if (tpkt->recv.headerCount >= TPKT_HEADER_SIZE)
            {
                int tpktLen=(((int)(tpkt->recv.header[2])<<8)+tpkt->recv.header[3])-TPKT_HEADER_SIZE;

                /* Do the actual reading */
                if (tpktLen > 0)
                {
                    if (socketRecvTcpIfConnected(tpkt, &finished) < 0)
                    {
                        int numOfLocks;

                        /* On error, close th eread event and notify */
                        tpkt->event= (RvSelectEvents) (tpkt->event & ~RvSelectRead);
                        RvSelectUpdate(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

                        numOfLocks = emaPrepareForCallback((EMAElement)tpkt);
                        tpkt->recv.eventHandler((HTPKT)tpkt,selectEvent,0,RV_TRUE,tpkt->recv.context);
                        emaReturnFromCallback((EMAElement)tpkt, numOfLocks);
                        break;
                    }
                }
                else
                {
                    /* finish the reading, we got a corrupted header */
                    tpktLen  = 0;
                    finished = RV_TRUE;
                }

                /* if we read all that we asked */
                if (finished)
                {
                    int numOfLocks;
                    /* Close the read event, notify how many bytes were read */
                    tpkt->recv.hRpool = NULL;
                    tpkt->event= (RvSelectEvents) (tpkt->event & ~RvSelectRead);
                    RvSelectUpdate(globals->pSelEngine, &tpkt->connection.fd, tpkt->event, tpktEvent);

                    numOfLocks = emaPrepareForCallback((EMAElement)tpkt);