tcppsx.c

extremeDB s sample code,useful for you

      }

      rc = select (1+(int)chan->socket, &readfds, 0, 0, ptv);

      if( rc < 0 ) // error
      {
        if(errno == EINTR) continue;
        goto ErrRet;
      }

      if( !rc ) // timeout
        goto ErrRet;

      if ( !(FD_ISSET((SOCKET)chan->socket, &readfds) ) )
   		  goto ErrRet;
      break;
    }    
    chan->status = NWST_INITIALIZED | NWST_CONNECTED;

    if(nw_recv(chan, (PCHAR)&msg, sizeof(conn_msg_t),
      (PSHORT)&len, timeout)!=MCO_S_OK) {
        goto ErrRet;
    }
    chan->index = msg.index;
    return MCO_S_OK;
}


/***************************************************************************
 *
 * Accept the cancel point connection
 *
 * Parameters:
 *
 * IN  nw_channel_h chan      - pointer to the listener channel.
 * IN OUT nw_channel_h  ioch  - pointer to the IO channel waiting for connection.
 * IN  unsigned long timeout) - wait-for-connect timeout.
 * 
 * Description:
 * 
 * Waits for the connection of a remote host to the IO channel.
 *
 * Return:
 *
 * Returns MSO_S_OK if successful or error code (see above).
 */
MCO_RET nw_accept_cancel_point(
                   nw_channel_h chan,
                   char * port,
                   timer_unit timeout
                   )
{
  int             csin_len;
  int             res;
  int             optval  = 1;
  SOCKADDR_IN     sin;          /* socket address structure, keeps port & address
                                  of the connection */
/*
 * Create listener TCP socket.
 */
    if ( (chan->socket = socket(PF_INET, SOCK_STREAM, 0)) < 0 )  {
      perror("\ncan't create listener socket ");
ErrRet:
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_ACCEPT);
    }

    setsockopt (chan->socket, SOL_SOCKET, SO_REUSEADDR,
      (const char*)&optval, sizeof(int) );
    
/* Set the socket to the listener state. */

    sin.sin_family      = AF_INET;
    sin.sin_addr.s_addr = htonl(INADDR_ANY);
    sin.sin_port        = htons((int2)atoi( port ));

    if ( bind(chan->socket, (LPSOCKADDR)&sin, sizeof(sin)) == -1 )
    {
      perror("\nbind error ");
      goto ErrRet;
    }

    if ( listen((SOCKET)chan->socket, SOMAXCONN) == -1 )
    {
      perror("\nlisten error ");
      goto ErrRet;
    }

/* Accept the connection */

    csin_len = sizeof(sin);
    
    if(
        (chan->cancel_socket =
          accept( chan->socket,
                  (LPSOCKADDR)&sin,
                  &csin_len
                )
        ) == -1
      )
    {
      perror("\naccept error ");
      goto ErrRet;
    }
    res = 1;

    setsockopt (chan->cancel_socket, IPPROTO_TCP, TCP_NODELAY,
      (const char*)&res, sizeof(int) );

    close(chan->socket);
    chan->socket = INVALID_SOCKET_VALUE;

    return MCO_S_OK;
}


/***************************************************************************
 *
 * Connect to cancel point
 *
 * Parameters:
 *
 * IN nw_channel_h   chan       - pointer to a channel descriptor
 * IN const char connect_string - interconnect dependent connection-string
 * IN  unsigned long timeout);
 * 
 * Description:
 * 
 * Connects IO channel to the remote host by it's name.
 *
 * Return:
 *
 * 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
                               )
{
  const char *cc = connect_string;
  char            name[128];
  int             len;
  struct hostent  *he;
  int             optval  = 1;
  int             res;
  SOCKADDR_IN     sin;          /* socket address structure, keeps port & address
                                  of the connection */

/* Translate interconnect-dependent connect-string
 * Format: "xxx.xxx.xxx.xxx:nnnn" or "inet-name:port" */
/* Find ':' in connect string */

    for (;;) {
      if ( *cc == '\0' ) return (MCO_RET)(MCO_E_NW_INVADDR);
      if ( *cc == ':' ) break;
      cc++;
    }

    cc++;
    len = cc-connect_string;

    if ((int)(cc-connect_string) >= (int)sizeof(name) ) {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_NOMEM);
    }
    strncpy(name, connect_string, (int)(cc-connect_string-1));
    name[(int)(cc-connect_string-1)] = 0;
/*
 * Translate IP address
 */
    if ( (he = gethostbyname(name)) == NULL ) return (MCO_RET)(MCO_E_NW_INVADDR);

    memcpy((char *) &sin.sin_addr, (char*) he->h_addr_list[0],
           (size_t) he->h_length);

    sin.sin_family = AF_INET;
    sin.sin_port = htons((uint2)atoi(cc));
  
/* Create  TCP socket. */
 
    if ( (chan->cancel_socket = socket(PF_INET, SOCK_STREAM, 0)) < 0 )  {
ErrRet: nw_close(chan);
        return (MCO_RET)(MCO_E_NW_CONNECT);
    }

    setsockopt (chan->cancel_socket, SOL_SOCKET, SO_REUSEADDR,
      (const char*)&optval, sizeof(int) );

    res = connect((SOCKET)chan->cancel_socket, (LPSOCKADDR)&sin, sizeof(sin));
    if( res == -1) {
      goto ErrRet;
    }

    return MCO_S_OK;
}

/***************************************************************************
 *
 * Send a message
 *
 * Parameters:
 *
 * IN nw_channel_h  chan    - pointer to a channel descriptor.
 * 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 a message to the communication channel 
 *
 * 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)
{
  int             rcs;
  fd_set          writefds;
  TIMEVAL         tv, *ptv = NULL;
  timer_unit      time = timeout;
  timer_unit      t1, t2;

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

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

    while( buflen )
    {

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

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

        tv.tv_sec = time/1000;
        tv.tv_usec= (time%1000)*1000;
      }

      rcs = select (1+(int)chan->socket , 0, &writefds, 0, ptv);

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

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

      if ( (rcs=send(chan->socket, buffer, buflen, 0)) <= 0 )
      {
        nw_close(chan);
        return (MCO_RET)(MCO_E_NW_RECVERR);
      }

      buflen = (short)(buflen - (int2) rcs);
      NumBytes += rcs;
      buffer += rcs;
    }

    return MCO_S_OK;
}

/***************************************************************************
 *
 * Receive a message
 *
 * 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 unsigned long timeout - receive timeout in milliseconds. It is media dependent
 *                            & application dependet. Application MUST set this parameter
 *                            considering the media rate & it's own needs
 *
 * Description:
 *
 * Receives a message from the communication cchannel
 *
 * 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;
  int2            length = buflen;
  fd_set          readfds;
  TIMEVAL         tv, *ptv = NULL;
  timer_unit      time = timeout;
  timer_unit      t1 = 0, t2;
  int 		  maxsel;

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

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

  while (length) {

    if(timeout != MCO_TM_INFINITE) {

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

      tv.tv_sec = time/1000;
      tv.tv_usec= (time%1000)*1000;
    }

    maxsel = chan->socket+1;
    FD_ZERO(&readfds);
    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( rcs < 0 )  { // error
      if( (errno == EINTR) || (errno == ENOSYS) ) continue;
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_RECVERR);
    }

    if( !rcs ) {  // timeout
      continue;
    }

    if(chan->cancel_socket != INVALID_SOCKET_VALUE) {
      if ( (FD_ISSET(chan->cancel_socket, &readfds) ) ) {
        recv(chan->cancel_socket, buffer, 4, 0);
#ifdef  NW_DEBUG_OUTPUT
        Printf("Replica was cancelled\n");
#endif
        return (MCO_RET)(MCO_E_NW_CANCEL);
      }
    }

    if ( !(FD_ISSET((SOCKET)chan->socket, &readfds) ) )    {
      continue;
    }

    if ( (rcs = recv(chan->socket, buffer, length, 0)) <= 0 )
    {
      nw_close(chan);
      return (MCO_RET)(MCO_E_NW_CLOSED);
    }
    length = (short)(length - (short)rcs);
    NumBytes += rcs;
    buffer += rcs;
  }

  *recvlen = (int2)buflen;
  return MCO_S_OK;

}


/************************************************************************
 *
 * 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_TCP_SOCKET_CHANNEL

#endif /* _WIN32 */