rpcsrv.c

eXtremeDB数据库在Linux平台下的使用源码。内含多个示例程序

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>

#ifdef _VXWORKS
#include <ioLib.h>
#define MSG_NOSIGNAL 0
#endif

#ifndef WIN32
#include <unistd.h>
#endif

#include "rpcsds.h"
#include "test_intf.h"
#include "test_intf_server.h"

#include <sys/types.h>

#ifdef _SOLARIS
#include <stropts.h>
#endif
 

#define DATA_THROUGH_A_NETWORK

//#define DATA_THROUGH_A_FILE

#ifdef DATA_THROUGH_A_FILE
typedef struct tag_test_params {
	FILE * w;
	FILE * r;
} test_params_t, *test_params_p;

int test_intf_write_stream ( void * buf_, unsigned int buf_sz, void * param, unsigned int network_order ) {

	char * buf = buf_;
	unsigned int i;
	for ( i=0; i<buf_sz; i++ ) {
		printf("%02hhx ", buf[i] );
	};
	printf("\n");
	
	fwrite( buf_, 1, buf_sz, ((test_params_p)param)->w );
	
	return 0;
};

int test_intf_read_stream ( void * buf_, unsigned int buf_sz, void * param, unsigned int network_order, unsigned int * read_sz ) {

	char * buf = buf_;
	unsigned int i;
	
	*read_sz = fread( buf_, 1, buf_sz, ((test_params_p)param)->r );

	for ( i=0; i<buf_sz; i++ ) {
		printf("%02hhx ", buf[i] );
	};
	printf("\n");
	
	return 0;
};

int test_intf_is_data_available( mco_rpc_context_p ctx, unsigned int * result ) {

/*
	char buf[100];
	printf( "mco_rpc_is_data_available?" ); fgets( buf, sizeof(buf), stdin );
*/
	
	return 0;
};
#endif

#ifdef DATA_THROUGH_A_NETWORK

#ifndef WIN32
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <netinet/tcp.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#if defined(_HPUX) || defined(_SOLARIS) || defined(_QNX)
#define MSG_NOSIGNAL 0
#endif
#else
#include <winsock2.h>
#define MSG_NOSIGNAL 0
#endif

#define CONNECTION_POOL	5
#define CONNECTION_ACCEPT_TIMEOUT_SEC  0
#define CONNECTION_ACCEPT_TIMEOUT_USEC 300*1000

typedef struct tag_test_params {
	
	int           tl   [CONNECTION_POOL+1];
	unsigned int  data_connection_i;
	
} test_params_t, *test_params_p;

int mco_init_network_server_interface( test_params_p nctx, char * intf, unsigned short port ) {

	unsigned int       i;
	struct sockaddr_in sin;

#ifdef WIN32
	WSADATA wsa; 
	int r = WSAStartup( MAKEWORD(2,0) , &wsa);
	if ( r != 0 ) {
		WSACleanup();
		return 1;
	};
#endif

	
	nctx->data_connection_i = 0;
	nctx->tl[0]             = socket( PF_INET, SOCK_STREAM, IPPROTO_TCP );
	
	for ( i=1; i<sizeof(nctx->tl)/sizeof(nctx->tl[0]); i++ ) {
		nctx->tl   [i] = -1;
	};

	/* make the first socket to listen for incoming connections */
	sin.sin_family      = AF_INET;
	sin.sin_port        = htons( port );
	sin.sin_addr.s_addr = inet_addr( intf );

	if ( bind  ( nctx->tl[0], (struct sockaddr*)&sin, sizeof (sin)) == -1 ) return 1;
	
	if ( listen( nctx->tl[0], SOMAXCONN ) == -1 ) return 2;

	return 0;
};

int mco_clear_network_server_interface( test_params_p nctx ) {

	unsigned int i;

	for ( i=0; i<sizeof(nctx->tl)/sizeof(nctx->tl[0]); i++ ) {

		if ( nctx->tl[i] >= 0 ) {
#ifndef WIN32  
#ifdef _VXWORKS
			shutdown( nctx->tl[i], 0 );
#else
			shutdown( nctx->tl[i], SHUT_RDWR );
#endif
			close   ( nctx->tl[i] );
#else /* WIN32 */
			shutdown( nctx->tl[i], SD_BOTH );
			closesocket ( nctx->tl[i] );
#endif /* WIN32 */
		};
	};

#ifdef WIN32  
	WSACleanup();
#endif /* WIN32 */
	
	return 0;
};

int test_intf_is_data_available( mco_rpc_context_p ctx, unsigned int * result ) {

	unsigned int     i, j;
	test_params_p    nctx = (test_params_p)ctx->param;
	struct timeval   tv;
	fd_set           fds;
  int              res;
  int              maxsockval = 0;	
  struct sockaddr_in  saddr;

	*result = 0;

    saddr.sin_family      = AF_INET; 
    saddr.sin_addr.s_addr = INADDR_ANY;
    saddr.sin_port        = htons(8082);

	tv.tv_sec  = CONNECTION_ACCEPT_TIMEOUT_SEC;
	tv.tv_usec = CONNECTION_ACCEPT_TIMEOUT_USEC;


/* zeroing fd's */
	FD_ZERO( &fds );
	
/* setting the open sockets */
  for ( i = 0; i<sizeof(nctx->tl)/sizeof(nctx->tl[0]); i++ ) {

    if ( nctx->tl[i] >= 0 ) {

      FD_SET( (unsigned int)nctx->tl[i], &fds );
      if(nctx->tl[i] > maxsockval)
      maxsockval	= nctx->tl[i];
		}

	}


/* wait for events on open sockets */
  while( (res = select ( maxsockval+1, &fds, 0, 0, &tv )) < 0) {
    if(errno != EINTR) 
      break;
  }

/* return if error or timeout */
  if( res <= 0) {
    return res;
  }

/* request sockets */
  for ( i =  0; i<sizeof(nctx->tl)/sizeof(nctx->tl[0]); i++ ) {
		unsigned int rcvd = 0;

    if ( nctx->tl[i] < 0 )
      continue;

    if ( !(FD_ISSET(nctx->tl[i], &fds) ) )
      continue;

    if (i == 0) {
/* listener socket */
/* find free element of the array & accept the connection */
    	for ( j = 1; j < sizeof(nctx->tl)/sizeof(nctx->tl[0]); j++ ) {
        if(nctx->tl[j] < 0) {
          res = sizeof(saddr) ;
		      nctx->tl[j] = accept(
                            nctx->tl[0], 
                            (struct sockaddr*)&saddr, 
                            &res
                            );
          break;
        }
      }
    }
    else {

#ifdef WIN32  
			res = ioctlsocket( nctx->tl[i], FIONREAD , &rcvd);
#else /* WIN32 */
#ifdef    _SOLARIS
			res = ioctl( nctx->tl[i], I_NREAD , &rcvd );
#else
			res = ioctl( nctx->tl[i], FIONREAD , &rcvd );
#endif
#endif
#ifdef    _SOLARIS
			if ( res ) {
#else
			if ( res == 0 ) {
#endif
			
				nctx->data_connection_i = i;
				*result = 1;
      	return 0;
			};

    }

  }

	return 0;
};

int test_intf_write_stream ( void * buf_, unsigned int buf_sz, void * param, unsigned int network_order ) {

	test_params_p   nctx = (test_params_p)param;
	char          * buf = buf_;
	unsigned int    i;
	
	if ( nctx->data_connection_i == 0 ) return 1;
	
	for ( i=0; i<buf_sz; i++ ) {
		printf("%02hhx ", buf[i] );
	};
	printf("\n");
	
	if ( -1 == send( nctx->tl[nctx->data_connection_i], buf_, buf_sz, MSG_NOSIGNAL ) ) return 1;
	
	return 0;
};

int test_intf_read_stream ( void * buf_, unsigned int buf_sz, void * param, unsigned int network_order, unsigned int * read_sz ) {

	test_params_p   nctx = (test_params_p)param;
	char          * buf = buf_;
	unsigned int    i;

	if ( -1 == ((*read_sz) = recv( nctx->tl[nctx->data_connection_i], buf_, buf_sz, MSG_NOSIGNAL )) ) return 1;

	for ( i=0; i<buf_sz; i++ ) {
		printf("%02hhx ", buf[i] );
	};
	printf("\n");
	
	return 0;
};

#endif

void mco_rpc_fatal_error( int e ) {

	printf( "Fatal error: %d\n", e );
	exit(1);
};

extern mco_rpc_context_t test_intf_ctx;

int test_int ( int i ) {
	
	return i | 0x5555AAAA;
};

void test_void_int ( int i ) {
	
};

int test_int_string ( char * s ) {
	
	return atoi(s);
};

int test_int_int5 ( int5 ints ) {
	
	return ints[0] + ints[1] + ints[2] + ints[3] + ints[4] ;
};

int test_int_pstruct( test_struct_p pstruct ) {
	
	return pstruct->a;
};

int test_int_variable_len( test_struct_variable_p p ) {
	return p->a_sz + p->b_len;
};

int test_int_union( test_struct_union_p p ) {

	return p->tp;
};

int hit_test() {

	struct timeval   tv;
	fd_set           fds;

	tv.tv_sec  = 0;
	tv.tv_usec = 1000;
	
	FD_ZERO( &fds );
	FD_SET( 0, &fds );
			
	return select ( 1, &fds, 0, 0, &tv ) > 0 ? 1 : 0; 
};

int main (int argc, char * argv[] ) {

	int r = 0;
	int test_Mode=0;
	test_params_t params;
	test_intf_ctx.param = &params;
#ifndef WIN32
	if (argc>1 && strstr(argv[1],"--test")) test_Mode=1;
#endif //

#ifdef DATA_THROUGH_A_FILE
	params.r = fopen( "test1_w.bin", "r" );
	params.w = fopen( "test1_r.bin", "w" );

	while ( r == 0 ) {
		printf( "%d\n", r = test_intf_dispatch( ) );
	};

	if ( params.w ) fclose( params.w );
	if ( params.r ) fclose( params.r );
#endif

#ifdef DATA_THROUGH_A_NETWORK
	if ( 0 != mco_init_network_server_interface ( &params, "0.0.0.0", 8082 ) ) {
		printf( "Unable to start the server\n" );
		return 1;
	};
	
	while ( test_Mode ||
		(r == 0 && 0 == hit_test() )) {
		printf( "%d\n", r = test_intf_dispatch( ) );
	};
	
	mco_clear_network_server_interface( &params );
#endif
	
	return 0;
};