udpw32.c

extremeDB s sample code,useful for you

 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_connect_cancel_point( nw_channel_h chan,
                                 const char * connect_string,  
                                 timer_unit timeout
                               )
{
    return MCO_S_OK;
}

/*************************************************************************
 * 
 * Description:
 *
 * Low level data transfer.
 * Sends a block of data to the communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h  chan    - pointer to a channel descriptor.
 * IN char*  buffer         - buffer to send.
 * IN int2 buflen           - buffer length,is cut to 32767 bytes.
 * IN timer_unit timeout    - send timeout in milliseconds. It is media dependent
 *                            & application dependent. Application MUST set this 
 *                            parameter considering the media rate & it's own needs.
 * 
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 *
 *************************************************************************/

MCO_RET nw_send ( 
                 nw_channel_h chan,  
                 char* buffer,
                 int2 buflen,   
                 timer_unit timeout
                )
{
  fd_set        writefds;
  TIMEVAL       tv, *ptv = NULL;
  int           rcs;
  SOCKADDR_IN   sin;
  int           len;
  short         count = buflen;
  timer_unit    time = timeout;
  timer_unit    t1, t2;
  
    if(timeout != MCO_TM_INFINITE) {
      t1  = t2  = mco_system_get_current_time();
      ptv = &tv;
    }

    if( (chan->status&(NWST_INITIALIZED|NWST_CONNECTED)) !=  (NWST_INITIALIZED|NWST_CONNECTED) ) {
      return (MCO_RET)(MCO_E_NW_CONNECT);
    }

    for(;;) {

      if(timeout != MCO_TM_INFINITE) {
        if( ((t2=mco_system_get_current_time())-t1) > time) {
          return (MCO_RET)(MCO_E_NW_TIMEOUT);
        }
        time -= t2 - t1;
        t2 = t1;
        
        tv.tv_sec = time/1000;
        tv.tv_usec= (time%1000)*1000;
      }

      FD_ZERO(&writefds);
      FD_SET(chan->socket, &writefds);

      rcs = select (0, 0, &writefds, 0, ptv);

      if ( rcs <= 0 )  {// error
	      nw_close(chan);
	      return (MCO_RET)(MCO_E_NW_SENDERR);
      }

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

      sin.sin_family      = AF_INET;             // Internet protocol stack
      sin.sin_addr.s_addr = htonl(chan->destaddr);
      sin.sin_port        = htons(chan->destport);
      len = sizeof(sin);

      if((len=sendto(chan->socket, buffer, count, 0, (LPSOCKADDR)&sin, len)) <=0 ) {
        nw_close(chan);
        return (MCO_RET)(MCO_E_NW_RECVERR);
      }

      count = (short)(count - len);
      NumBytes += len;
      buffer   += len;
      return MCO_S_OK;
    }
}


/************************************************************************
 *
 *  Description:
 *
 * Low level data transfer.
 * Receives block of data from the communication channel
 *
 *  Parameters:
 *
 * IN  nw_channel_h chan  - pointer to a channel descriptor.
 * OUT char* buffer       - buffer to receive.
 * IN  int2 buflen        - buffer length limit,  is cut to 32767 bytes.
 * OUT int2* recvlen,     - actual received length, is cut to 32767 bytes.
 * IN  timer_unit timeout - receive timeout in milliseconds. It is media dependent
 *                            & application dependet. Application MUST set this parameter
 *                            considering the media rate & it's own needs
 *  Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 *
 ************************************************************************/

MCO_RET nw_recv( 
                nw_channel_h  chan,  
                char*         buffer,  
                int2          buflen,
                int2*         recvlen,  
                timer_unit    timeout
               )
{
  int             rcs;
  WSAOVERLAPPED   ovl;
  WSABUF          wsabuf;
  uint4           flags;
  uint4           st;
  int             len;
  timer_unit    time = timeout;
  timer_unit      t1, t2;

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

     for(;;) {

       if(timeout != MCO_TM_INFINITE) {

         if( ((t2=mco_system_get_current_time())-t1) > time) {
           return (MCO_RET)(MCO_E_NW_TIMEOUT);
         }
         time -= t2 - t1;
         t2 = t1;
       }

       wsabuf.buf = (PCHAR)buffer;
       wsabuf.len = (uint4)buflen;
       ovl.hEvent = chan->event;
       flags  = 0;
       rcs = buflen;
       len = sizeof(chan->sin);

       if (
          WSARecvFrom (
            chan->socket,
            &wsabuf,
            1,
            (LPDWORD)&rcs,
            (LPDWORD)&flags,
            (struct sockaddr *)&chan->sin,
            (LPINT)&len,
            &ovl,
            NULL
          )
            < 0
         ) 
       {
          int err;
          if((err = WSAGetLastError())!=WSA_IO_PENDING) {
            return (MCO_RET)(MCO_E_NW_RECVERR);
          }

          if((st =WSAWaitForMultipleEvents(2,(const WSAEVENT FAR *)&chan->event, 0, time, 0))
           != WSA_WAIT_EVENT_0 ) {
             if(st == WSA_WAIT_FAILED) {
              nw_close(chan);
              return (MCO_RET)(MCO_E_NW_FATAL);
             }
             if(st == WSA_WAIT_EVENT_0+1) {
#ifdef  NW_DEBUG_OUTPUT
              Printf("Replica was cancelled\n");
#endif
              nw_close(chan);
              return (MCO_RET)(MCO_E_NW_CANCEL);
             }
            return (MCO_RET)(MCO_E_NW_TIMEOUT);
          }
          WSAGetOverlappedResult( chan->socket, &ovl,
                             (PULONG)&rcs, 0, (PDWORD)&flags );

      }

      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;
    }
}


/******************************************************************************
 * 
 * Description:
 * 
 * Sends message of any type to a communication channel
 *
 * 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 timer_unit timeout     - send timeout in milliseconds. It is media dependent
 *                            & application dependent. Application MUST set this 
 *                            parameter considering the media rate & it's own needs.
 *
 * 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;
}

/******************************************************************************
 *
 * Description:
 * Receive message of certain type from a communication channel
 *
 * 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 timer_unit timeout  - receive timeout
 *
 * 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,
                         PSHORT       recvlen,   
                         timer_unit   timeout)
{
  int2            repeat_count = REPEAT_COUNT;
  uint2           rest = (uint2)buflen;
  uint2           len;
  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((PVOID)buffer, chan->pBuffer, (int)len);
        chan->pBuffer   += len;
        (PUCHAR)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;
          }
          break;
        }

        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 = (PUCHAR)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((PVOID)buffer, chan->pMsg->buffer, len);

        chan->pBuffer   += len;
        (PUCHAR)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);
}

/************************************************************************
 *
 * Cancels the communication channel
 *
 * Parameters:
 *
 * IN nw_channel_h ch      - pointer to a channel descriptor.
 *
 */
void  nw_cancel(nw_channel_h ch)
{
  ch->status |= NWST_CANCEL;
  SetEvent(ch->event1 );
}

#endif// CFG_UDP_SOCKET_CHANNEL

#endif /* _WIN32 */