udpvx.c

extremeDB s sample code,useful for you

      FD_SET((SOCKET)chan->socket, &readfds);

      if(chan->cancel_socket != INVALID_SOCKET_VALUE) {
        FD_SET((SOCKET)chan->cancel_socket, &readfds);
        if(chan->cancel_socket > chan->socket)
          maxsel = chan->cancel_socket+1;
      }

      rcs = select (maxsel, &readfds, 0, 0, ptv);

      if(chan->status & NWST_CANCEL) {
       return (MCO_RET)(MCO_E_NW_CANCEL);
      }

      if( rcs < 0 ) // error
      {
      if(errno == EINTR) continue;
       nw_close(chan);
       return (MCO_RET)(MCO_E_NW_RECVERR);
      }

      if ( !rcs ) {
       return (MCO_RET)(MCO_E_NW_TIMEOUT);
      }

      if(chan->cancel_socket != INVALID_SOCKET_VALUE) {
        if ( (FD_ISSET(chan->cancel_socket, &readfds) ) ) {
          recv(chan->cancel_socket, buffer, 4, 0);
          return (MCO_RET)(MCO_E_NW_CANCEL);
        }
      }

      if ( !(FD_ISSET(chan->socket, &readfds) ) )  {
        nw_close(chan);
        return (MCO_RET)(MCO_E_NW_RECVERR);
      }

      len=sizeof(chan->sin);

      if (
           (rcs=
              recvfrom(
                       chan->socket,
                       buffer,
                       buflen,
                       0,
                       (LPSOCKADDR)&chan->sin,
                       &len)
           ) < 0 
         )
      {
        if(errno==EMSGSIZE) continue;
        nw_close(chan);
        return (MCO_RET)(MCO_E_NW_RECVERR);
      }
      chan->inaddr = ntohl(chan->sin.sin_addr.s_addr);  // source address
      chan->inport = ntohs(chan->sin.sin_port);
      NumBytes += rcs;
      *recvlen = (int2)rcs;
      return MCO_S_OK;
   }
}

/******************************************************************************
 *
 * Send guarrantied message to a communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h  chan    - pointer to channel descriptor.
 * IN uint1 type            - message type
 * IN  const char*  buffer  - buffer to send.
 * IN int2 buflen           - buffer length,is cut to 32767 bytes.
 * IN  unsigned long timeout - send timeout in milliseconds. It is media dependent
 *                            & application dependent. Application MUST set this
 *                            parameter considering the media rate & it's own needs.
 *
 * Description:
 *
 * Sends guarrantied message of any type to a communication channel
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET _nw_send_message(
                nw_channel_h chan,
                uint1        type,
                const void*  buffer,
                int2         buflen,
                timer_unit   timeout)
{
  int2            repeat_count = REPEAT_COUNT;
  int2            len;
  int2            size;
  MCO_RET         rc= (MCO_RET)(0);
  udp_header_h    pHeader;
  timer_unit      time   = timeout;
  timer_unit      iotime = timeout;
  timer_unit      t1, t2;

    if(timeout != MCO_TM_INFINITE) {
      iotime /= REPEAT_COUNT;
      t1  = t2  = mco_system_get_current_time();
    }

    while(buflen) {

      if( (size = buflen) > MSG_BUFFER_SIZE )
        size = MSG_BUFFER_SIZE;

      pHeader= (udp_header_h)(&chan->pMsg->hd);
      pHeader->magic  = MAGIC;
      pHeader->ctrl   = (uint1)((chan->send_seqn << FC_SEQN_OFFS) | type); // init control byte

      if((pHeader->length=(uint2)size) != 0) memcpy(chan->pMsg->buffer,buffer,size);


      /* try to send message REPEAT_COUNT times */
      for( repeat_count = REPEAT_COUNT; repeat_count; repeat_count--) {

        if(timeout != MCO_TM_INFINITE) {
          if( ((t2=mco_system_get_current_time())-t1) > time) {
            nw_close(chan);
            return (MCO_RET)(MCO_E_NW_TIMEOUT);
          }
          time -= (t2 - t1);
          t2 = t1;
        }

        /* send message  */
        if(
            (rc =
              nw_send (
                  chan,
                  (char*)chan->pMsg,
                  (short)(size+sizeof(udp_header_t)),
                  iotime)
            ) != MCO_S_OK
          ) break;

        /* receive an acknowlege */
        if (
             (rc =
                nw_recv(
                      chan,
                      (char*)&chan->header,
                      sizeof(udp_header_t),
                      &len,
                      iotime)
             ) //RCV_ACK_TIME  
             != MCO_S_OK
           )
        {
          if( rc != MCO_E_NW_TIMEOUT) break;
          continue;
        }

        /* check frame length */
        rc = (MCO_RET)(MCO_E_NW_ERROR);
        if(chan->header.magic != MAGIC) {
          /* acknowledge header is broken, try once more */
          continue;
        }
        if(len != sizeof(udp_header_t)) {
          continue; 
        }
        if(chan->header.length) {
          continue; 
        }
        /* check if it is the acknowlege of corresponding type of a frame */
        if( (chan->header.ctrl&FC_TYPE_MASK) != type+1) {
          continue;
        }
        /* check sequence number, reject if repeated */
        if( ((chan->header.ctrl&FC_SEQN_MASK) >> FC_SEQN_OFFS)
          != chan->send_seqn) {
            continue;
        }

        /* increase send sequence number */

        chan->send_seqn = (uint1)((++chan->send_seqn) & (FC_SEQN_MASK >> FC_SEQN_OFFS));
        rc = MCO_S_OK;
        break;
      }

      if(rc != MCO_S_OK) {
        nw_close(chan);
        break;
      }
      buflen        -= size;
      (char*)buffer += size;

    }
    return rc;
}

/******************************************************************************
 *
 * Receive guarrantied message from a communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h chan   - pointer to a channel descriptor
 * IN uint1 type          - message type
 * IN void* buffer        - receive buffer
 * IN uint2 buflen        - receive buffer limit
 * OUT uint2* recvlen     - return actual length
 * IN long timeout        - receive timeout
 *
 * Description:
 *
 * Receives guarrantied message of certain type from a communication channel
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET _nw_recv_message(
                nw_channel_h  chan,
                uint1         type,
                const void*   buffer,
                int2          buflen,
                int2*         recvlen,
                timer_unit    timeout)
{
  int2          repeat_count = REPEAT_COUNT;
  int2          len;
  uint2         rest = (uint2)buflen;
  MCO_RET       rc=(MCO_RET)(0);
  uint1         seqn;
  udp_header_h  pHeader;
  timer_unit    time    = timeout;
  timer_unit    iotime  = timeout;
  timer_unit    t1, t2;

    if(timeout != MCO_TM_INFINITE) {
      iotime /= REPEAT_COUNT;
      t1  = t2  = mco_system_get_current_time();
    }

    while(rest != 0) {

      if(chan->currlen != 0) {

        rc=(MCO_RET)(0);
        len = chan->currlen;
        if(len > rest) len = rest;

        memcpy((void*)buffer, chan->pBuffer, (int)len);
        chan->pBuffer   += len;
        (char*)buffer   += len;
        chan->currlen   -= (uint2)len;
        rest -= len;
        continue;
      }

      /* attempt to receive message REPEAT_COUNT times */
      for( repeat_count = REPEAT_COUNT; repeat_count; repeat_count--) {

        if(timeout != MCO_TM_INFINITE) {
          if(((t2=mco_system_get_current_time())-t1) > time) {
            nw_close(chan);
            return (MCO_RET)(MCO_E_NW_TIMEOUT);
          }
          time -= t2 - t1;
          t1 = t2;
        }

        /* receive message */
        if (
             (rc =
                nw_recv(
                     chan,
                     (char*)chan->pMsg,
                     (int2)(chan->bufsize),
                     &len,
                     iotime)
             ) != MCO_S_OK
           )
        {
          if(rc != MCO_E_NW_TIMEOUT) {
              nw_close(chan);
              return rc;
          }
          continue;
        }

        rc = (MCO_RET)(MCO_E_NW_ERROR);
        if(((pHeader=(udp_header_h)(&chan->pMsg->hd)))->magic != MAGIC) {
          continue;                    // the header is broken, wait for the next
        }
        if(len != (int2)(pHeader->length+sizeof(udp_header_t))) {
            return (MCO_RET)(MCO_E_NW_ERROR); // invalid data length
        }

        /* if message type not corresponds then wait for next message */
        if((pHeader->ctrl&FC_TYPE_MASK) != type ) {
          continue;
        }

        /* prepare the acknowlege header */
        memcpy(&chan->header,pHeader,sizeof(udp_header_t));
        chan->header.length = 0;
        chan->header.ctrl++; // set acknowlege type
        if(!chan->destaddr) {
          chan->destaddr = chan->inaddr;
          chan->destport = chan->inport;
        }

        if(timeout != MCO_TM_INFINITE) {
          if( ((t2=mco_system_get_current_time())-t1) > time) {
            nw_close(chan);
            return (MCO_RET)(MCO_E_NW_TIMEOUT);
          }
          time -= t2 - t1;
          t2 = t1;
        }

        /* send acknowlege of corresponding type */
        if(
            (rc = 
              nw_send(
                    chan,
                    (char*)&chan->header,
                    sizeof(udp_header_t),
                    iotime) //SEND_ACK_TIME
            ) != MCO_S_OK
          )
        {
            break;
        }

        /* check sequence number, if repeated then wait for next message*/
        if( (seqn=(uint1)((pHeader->ctrl&FC_SEQN_MASK) >> FC_SEQN_OFFS))
          != chan->recv_seqn) {
            rc = (MCO_RET)(MCO_E_NW_ERROR);
            continue;
        }

        chan->recv_seqn = (uint1)((++chan->recv_seqn) & (FC_SEQN_MASK >> FC_SEQN_OFFS));

        chan->currlen = (uint2)(len -= sizeof(udp_header_t));
        chan->pBuffer = (char*)chan->pMsg->buffer;

        len = chan->pMsg->hd.length;

        if( len > rest)
          len = rest;

        if( len > (int2)chan->currlen)
          len = chan->currlen;

        if( len != 0 ) memcpy((void*)buffer, chan->pMsg->buffer, len);

        chan->pBuffer   += len;
        (char*)buffer   += len;
        chan->currlen   -= (uint2)len;
        rest -= len;

        break;
      }

      if(rc != MCO_S_OK) {
        nw_close(chan);
        break;
      }
    }

    *recvlen=buflen;
    return  rc;
}

/******************************************************************************
 * wrapper functions nw_send_message(), nw_recv_message() - for sending/receiving
 * messages of type DATA
 */
MCO_RET nw_send_message( nw_channel_h chan,
                         const void*  buffer,
                         int2 buflen,
                         long timeout)
{
  return  _nw_send_message(chan, FT_DATA, buffer, buflen, timeout);
}

MCO_RET nw_recv_message( nw_channel_h chan,
                         const void* buffer,
                         int2 buflen,
                         int2* recvlen,
                         long timeout)
{
    return  _nw_recv_message(chan, FT_DATA, buffer, buflen, recvlen, timeout);
}

/************************************************************************\n
 *
 * Cancels the communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h ch      - pointer to a channel descriptor.
 *
 */
void  nw_cancel(nw_channel_h ch)
{
  uint4 data;

  ch->status |= NWST_CANCEL;
  send(ch->cancel_socket, &data, 4, 0);
}

#endif// CFG_UDP_SOCKET_CHANNEL