tpkt.c

基于h323协议的软phone

/***********************************************************************
        Copyright (c) 2002 RADVISION Ltd.
************************************************************************
NOTICE:
This document contains information that is confidential and proprietary
to RADVISION Ltd.. No part of this document may be reproduced in any
form whatsoever without written prior approval by RADVISION Ltd..

RADVISION Ltd. reserve the right to revise this publication and make
changes without obligation to notify any person of such revisions or
changes.
***********************************************************************/

#include <stdio.h>
#include <rvsocket.h>
#include "rvinternal.h"
#include "ema.h"
#include "rpool.h"
#include "cmutils.h"
#include "cm.h"
#include "rvwatchdog.h"
#include "transportint.h"
#include "transnet.h"
#include "tpkt.h"

#ifdef __cplusplus
extern "C" {
#endif

static char tpktDummyBuff[512];

/* TPKT instance */
typedef struct
{
    RvSelectEngine* pSelEngine; /* Select Engine, used for the file descriptors */
    RvPortRange*    pPortRange; /* Port range to use */
    HEMA            hEma; /* EMA used for TPKT elements */
    RvLogSource     log; /* Log source to use */
} tpktGlobals;

typedef struct
{
    LPTPKTEVENTHANDLER  eventHandler;               /* Callback routine on transaction event */
    RvUint8             header[TPKT_HEADER_SIZE];   /* TPKT header */
    void*               context;                    /* additional data to be sent on the callback */
    RvSize_t            headerCount;                /* an index into the buffer indicationg how much
                                                       of the header was processsed */
    HRPOOL              hRpool;                     /* the rpool where the message is */
    HRPOOLELEM          message;                   /* the message to be sent/read */
    int                 offset;                     /* the offset into the message */
    RvLock*             rpoolLock;                  /* the lock to protect the rpool */
} tpktTrans;

/* tpktInfo holds information about a single TPKT connection. This is held inside EMA */
typedef struct
{
    RvH323Connection    connection; /* The network connection */
    RvUint32            bytesLeftToRead; /* Number of bytes left to read on the connection */
    RvBool              readingHeader; /* RV_TRUE if we're reading a TPKT header, RV_FALSE if
                                          we're reading the message itself */
    RvBool              errorEncountered; /* RV_TRUE if we encountered an error while trying
                                             to read from this connection */ /* todo: handle this error */
    RvUint8             header[TPKT_HEADER_SIZE]; /* TPKT header of current incoming message */
    tpktTrans           recv;    /* The receiving transaction data */
    tpktTrans           send;    /* The sending transaction data */
    tpktTypes           type;    /* The type of connection (MultiServer, Server or Client */
    RvBool              isConnected;    /* The type of connection (MultiServer, Server or Client */
    RvSelectEvents      event;
    RvBool              close;
} tpktInfo;

#define socketSendTcpIfConnected(tpkt, finished) \
    (((tpkt)->isConnected)?(tpktSocketSendTcp(tpkt, (finished))):RV_OK)

#define socketRecvTcpDirectIfConnected(tpkt, buf, len, pLenRecv) \
    (((tpkt)->isConnected)?(RvSocketReceiveBuffer(&((tpkt)->connection.socket), (buf), (len), (pLenRecv), NULL)):RV_OK)

#define socketRecvTcpIfConnected(tpkt, finished) \
    (((tpkt)->isConnected)?(tpktSocketRecvTcp(tpkt, (finished))):RV_OK)


/*******************************************************************************************
 * tpktSocketSendTcp
 *
 * Purpose: internal routine that simulates RvSocketSendBuffer using rpool buffer rather than
 *          a contiguous one, handling the problems arising that.
 *
 * Input:   tpkt - The object of the send, having all the necessary data abour rpool
 *
 * Output: finished - whether the send was finished already or not
 *
 * Return Value: number of bytes sent.
 *******************************************************************************************/
static int tpktSocketSendTcp(tpktInfo* tpkt, RvBool *finished)
{
    int sent = 0, len, totalLength, totalSent = 0;
    RvUint8 *buffer;
    RvStatus ret;

    /* lock around operation with the rpool */
    RvLockGet(tpkt->send.rpoolLock);

    /* get the total length of the message */
    totalLength = rpoolChunkSize(tpkt->send.hRpool, tpkt->send.message);

    /* send contiguous segment at a time until all the message is sent or TCP is exhausted */
    do {
        /* get the current segment pointer and length */
        len = rpoolGetPtr(tpkt->send.hRpool,
                          tpkt->send.message,
                          tpkt->send.offset,
                          (void **)&buffer);

        /* do the actual send */
        ret = RvSocketSendBuffer(&tpkt->connection.socket, buffer, (RvSize_t)len, NULL, (RvSize_t*)&sent);

        /* update the pointer into the message and the current call counter */
        if ((ret == RV_OK) && (sent >= 0))
        {
            tpkt->send.offset += sent;
            totalSent += sent;
        }
        else
        {
            /* Error: we couldn't write even one segment */
            *finished = RV_FALSE;
            /* release rpool */
            RvLockRelease(tpkt->send.rpoolLock);
            /* report the error */
            return ret;
        }

        /* if TCP was exhausted before all we wanted was sent */
        if (sent < len)
        {
            /* we couldn't write even one segment, wait for event write */
            *finished = RV_FALSE;
            /* release rpool */
            RvLockRelease(tpkt->send.rpoolLock);
            /* report how many bytes were read in this call */
            return totalSent;
        }

    } while (tpkt->send.offset < totalLength);

    /* if we're here then all the message was sent */
    *finished = RV_TRUE;
    RvLockRelease(tpkt->send.rpoolLock);
    /* report that the whole message was sent */
    return totalSent;
}

/*******************************************************************************************
* tpktSocketRecvTcp
*
* Purpose: internal routine that simulates RvSocketReceiveBuffer using rpool buffer rather than
*          a contiguous one, handling the problems arising from that.
*
* Input:   tpkt - The object of the receive, having all the necessary data abour rpool
*
* Output: finished - whether the receive was finished already or not
*
* Return Value: number of bytes sent.
*******************************************************************************************/
int tpktSocketRecvTcp(tpktInfo* tpkt, RvBool *finished)
{
    int received = 0, len, totalLength, totalReceived = 0;
    RvUint8 *buffer;
    RvStatus status;

    /* lock around operation with the rpool */
    RvLockGet(tpkt->recv.rpoolLock);

    /* get the total length of the currently allocated rpool message
       This should be exactly the size of the message as was received in the header */
    totalLength = rpoolChunkSize(tpkt->recv.hRpool, tpkt->recv.message);

    /* 2004.12.04, added. fujiangdong. when there is no incoming message, do not call 
        RvSocketReceiveBuffer function, otherwise it will be pended here forever.
      */
    if (totalLength > 0)
		
    /* read one segment after the other until the whole message is
       read or TCP is empty for a while */
    do {
        /* get a pointer and length for the current segment */
        len = rpoolGetPtr(tpkt->recv.hRpool,
                          tpkt->recv.message,
                          tpkt->recv.offset,
                          (void **)&buffer);

        /* do the actual receive */
        status = RvSocketReceiveBuffer(&tpkt->connection.socket, buffer, (RvSize_t)len, (RvSize_t *)&received, NULL);
        /* update the offset into the rpool message and the current call counter */
        tpkt->recv.offset += received;
        totalReceived += received;
        /* if TCP was empty before we read all we wanted wait for next event */
        if ((status != RV_OK) || (received < len))
        {
            /* we couldn't read even one segment, wait for event write */
            *finished = RV_FALSE;

            /* release rpool */
            RvLockRelease(tpkt->recv.rpoolLock);

            /* If we had an error receiving, then this connection is most likely dead */
            if (status != RV_OK)
                return status;
            
            /* report how many bytes were read in this call */
            return totalReceived;
        }
    } while (tpkt->recv.offset < totalLength);

    /* here it means that the whole message was read */
    *finished = RV_TRUE;
    RvLockRelease(tpkt->recv.rpoolLock);
    return totalLength;
}

/*******************************************************************************************
 * tpktCloseElement
 *
 * Purpose: internal routine that closes one tpkt element. Used at shutdown to be called
 *          on all the remaining elements.
 *
 * Input:   elem - The tpkt element
 *          param - Dummy parameter.
 *
 *******************************************************************************************/
static void* tpktCloseElement(IN EMAElement elem, IN void* param)
{
    tpktInfo* tpkt = (tpktInfo *)elem;
    tpktGlobals* globals = (tpktGlobals*)emaGetUserData(elem);

    RV_UNUSED_ARG(param);

    if ((tpkt != NULL) && (globals != NULL))
    {
        tpkt->close = RV_TRUE;
        /* Close the connection */
        tpktClose((HTPKT)tpkt);
    }

    return NULL;
}

/*******************************************************************************************
 * tpktInit
 *
 * Purpose: Initialize the structures of the tpkt module
 *
 * Input:   hApp        - Stack's application handle
 *          selEngine   - Select Engine, used for the file descriptors
 *          sessions    - Maximal number of allowed connections
 *          portRange   - Range of ports to use
 * Return Value: Handle to the TPKT instance
 *******************************************************************************************/
HTPKTCTRL tpktInit(
    IN HAPP            hApp,
    IN RvSelectEngine* selEngine,
    IN int             sessions,
    IN RvPortRange*    portRange)
{
    tpktGlobals*    globals;

    /* allocate global area for tpkt package */
    if(RvMemoryAlloc(NULL, (void**)&globals, sizeof(tpktGlobals)) != RV_OK)
        return NULL;

    /* Build the EMA pool of tpktInfo elements */
    globals->hEma = emaConstruct(
        sizeof(tpktInfo), sessions + 1,
        emaNormalLocks, "TPKT elements", 0, globals, NULL);
    if (globals->hEma == NULL)
    {
        RvMemoryFree(globals);
        return NULL;
    }

    emaAddWatchdogResource(globals->hEma, cmiGetWatchdogHandle(hApp), "TpktElements", "Transport", "TPKT Connection elements");

    globals->pSelEngine = selEngine;
    globals->pPortRange = portRange;

    RvLogSourceConstruct(RvLogGet(), &globals->log, RV_LOG_LIBCODE_H323, "TCP", "Lower TPKT layer");

    return (HTPKTCTRL)globals;
}


/*******************************************************************************************
 * tpktEnd
 *
 * Purpose: Destroy the tpktInfo elements and close all connections
 *
 * Input:   hCtrl - Handle to the EMA of tpktInfo elements
 *
 * Return Value: 0
 *******************************************************************************************/
int tpktEnd(HTPKTCTRL hCtrl)
{
    tpktGlobals *globals = (tpktGlobals *)hCtrl;

    /* Go over all tpktInfo elements in the EMA pool and close them */
    emaDoAll(globals->hEma, tpktCloseElement, NULL);

    /* destroy the EMA pool of tpktInfo elements */
    emaDestruct(globals->hEma);

    RvLogSourceDestruct(RvLogGet(), &globals->log);

    /* release the global area of tpkt */
    RvMemoryFree(globals);

    return 0;
}


/*******************************************************************************************