mserver.mx

一个内存数据库的源代码这是服务器端还有客户端

}

@- Wrappers
The MonetDB Version 5 wrappers are collected here
The latest port known to gain access is stored
in the database, so that others can more easily
be notified.
@c
str
SERVERlisten_default(int *ret)
{
	int port = SERVERPORT;
	str p;
	int maxusers = SERVERMAXUSERS;

	(void) ret;
	p = GDKgetenv("mapi_port");
	if (p)
		port = (int) strtol(p, NULL, 10);
	return SERVERlisten(&port, 0, &maxusers, 0);
}

str
SERVERlisten2(int *ret, int *port, int *maxusers)
{
	(void) ret;
	return SERVERlisten(port, 0, maxusers, 0);
}

str
SERVERlisten3(int *ret, int *port, int *maxusers, str *cmd)
{
	(void) ret;
	return SERVERlisten(port, 0, maxusers, cmd);
}

str
SERVERlisten_port(int *ret, int *pid)
{
	int port = *pid;
	int maxusers = SERVERMAXUSERS;

	(void) ret;
	return SERVERlisten(&port, 0, &maxusers, 0);
}
@-
The internet connection listener may be terminated from the server console,
or temporarily suspended to enable system maintenance.
It is advisable to trace the interactions of clients on the server
side. At least as far as it concerns requests received.
The kernel supports this 'spying' behavior with a file descriptor
field in the client record.
@c

str
SERVERstop(int *ret)
{
	int i;
	for( i=0; i< lastlistener; i++)
		MT_kill_thread(listener[i]);
	lastlistener = 0;
	(void) ret;		/* fool compiler */
	return MAL_SUCCEED;
}


str
SERVERsuspend(int *res)
{
	(void) res;
	serveractive= FALSE;
	return MAL_SUCCEED;
}

str
SERVERresume(int *res)
{
	serveractive= TRUE;
	(void) res;
	return MAL_SUCCEED;
}

str
SERVERclient(int *res, stream **In, stream **Out) 
{
	int i = 0;
	char *name = (char *) GDKmalloc(BLOCK), *user = name, *s;
	char *challenge;
	int blocked = 0;
	bstream *bs;
	stream *fdin = *In, *fdout = *Out;

	(void) res;
	assert(name);
	memset(name, 0, BLOCK);

	/* is this routine used?  Where? */
	assert(0);

	/* write challenge string, the challenge is currently empty (disabled) */
	challenge = "::mserver_mal:4\n";

	/* temporarily misuse name */
	snprintf(name, BLOCK, "%2u%s", (unsigned int) strlen(challenge), challenge);
	stream_write(fdout, name, strlen(name), 1);
	/* get user name */
	*user = 0;
	stream_read(fdin, user, 1, 1);
	for (i = 0; !stream_errnr(fdin) && i < BLOCK && *user != '\n'; i++) {
		user++;
		stream_read(fdin, user, 1, 1);
	}
	if (stream_errnr(fdin)) {
		throw(IO, "SERVERclient", "fdin write problems");
	}
	*user = 0;
	/* in embedded mode we allow just one client */
#ifdef DEBUG_SERVER
	printf("mserver:Client accepted %s\n", name);
	fflush(stdout);
#endif
	s = strrchr(name, ':');
	blocked = s && strncmp(s, ":blocked", 8) == 0;
	if (blocked) {
		/* printf("mserver:use blocked mode\n");fflush(stdout); */
		*s = 0;
		fdin = block_stream(stream_rstream(fdin));
		fdout = block_stream(stream_wstream(fdout));
	}
#ifdef DEBUG_SERVER
	THRprintf(GDKout, "#SERVERlisten:client accepted %d\n", msgsock);
	THRprintf(GDKout, "#SERVERlisten:client string %s\n", name);
#endif
	bs = bstream_create(fdin, blocked ? 128 * BLOCK : 0);

	if (blocked)
		bs->eof = 1;
	MSscheduleClient(name, challenge, bs, fdout);
	return MAL_SUCCEED;
}

void
SERVERexit(){
	int ret;
	SERVERstop(&ret);
	/* remove any port identity file */
	system("rm -rf .*_port");
}
@+ Remote Processing
The remainder of the file contains the wrappers around
the Mapi library used by application programmers.
Details on the functions can be found there.

Sessions have a lifetime different from dynamic scopes.
This means the  user should use a session identifier
to select the correct handle.
For the time being we use the index in the global
session table. The client pointer is retained to
perform access control. 

We use a single result set handle. All data should be
consumed before continueing.

A few extra routines should be defined to
dump and inspect the sessions table.
	
The remote site may return a single error
with a series of error lines. These contain
then a starting !. They are all stripped here.
@= catchErrors
{int rn;
	if ( ( (rn=mapi_error(mid)) == -4 && hdl && mapi_result_error(hdl))  || rn){
		str err,newerr;
		size_t l;
		char *e,*f;

		if( hdl && mapi_result_error(hdl))
			err= mapi_result_error(hdl);
		else err= mapi_result_error(SERVERsessions[i].hdl);
		l= 2*strlen(err)+8192;
		newerr= (str) alloca(l);

		if (err == NULL)
			err = "(no additional error message)";
		f= newerr; 
		for( e=err+1; *e && l>1; e++)
			if( *e== '!' && *(e-1)=='\n'){
				snprintf(f,l,"MALException:@1:remote error:");
				l-= strlen(f);
				while(*f) f++;
			} else{
				*f++ = *e;
				l--;
			}
		*f=0;
		throw(MAL, "@1", "remote error: %s", newerr);
	}
}
@c
#define MAXSESSIONS 32
struct{
	int key;
	str dbalias;	/* logical name of the session */
	Client c;
	Mapi mid;		/* communication channel */
	MapiHdl hdl;	/* result set handle */
} SERVERsessions[MAXSESSIONS];

static int sessionkey=0;

/* #define MAPI_TEST*/

str
SERVERconnectAll(int *key, str *host, int *port, str *username, str *password, str *lang, int ssl){
	Mapi mid;
	int i;

	mal_set_lock(mal_contextLock,"SERVERconnect");
	for(i=1; i< MAXSESSIONS; i++)
	if( SERVERsessions[i].c ==0 ) break;

	if( i==MAXSESSIONS){
		mal_unset_lock(mal_contextLock,"SERVERconnect");
		throw(IO, "mserver.connect", "Too many sessions");
	}
	SERVERsessions[i].c= MCgetClient();
	SERVERsessions[i].key= ++sessionkey;
	mal_unset_lock(mal_contextLock,"SERVERconnect");

	if (ssl)
		mid = mapi_connect_ssl(*host, *port, *username, *password, *lang, NULL);
	else
		mid = mapi_connect(*host, *port, *username, *password, *lang, NULL);

	if (mapi_error(mid)) {
		str err = mapi_error_str(mid);
		if (err == NULL)
			err = "(no reason given)";
		if (err[0] == '!')
			err = err + 1;
		SERVERsessions[i].c = NULL;
		throw(IO, "mserver.connect", "Could not connect: %s", err);
	}

#ifdef MAPI_TEST
	stream_printf(GDKout,"Succeeded to establish session\n");
#endif
	SERVERsessions[i].mid= mid;
	*key = SERVERsessions[i].key;
	return MAL_SUCCEED;
}

str
SERVERdisconnectALL(int *key){
	int i;

	mal_set_lock(mal_contextLock,"SERVERdisconnect");

	for(i=1; i< MAXSESSIONS; i++)
		if( SERVERsessions[i].c != 0 ) {
			SERVERsessions[i].c = NULL;
			*key = SERVERsessions[i].key;
			mapi_disconnect(SERVERsessions[i].mid);
		}

	mal_unset_lock(mal_contextLock,"SERVERdisconnect");

#ifdef MAPI_TEST
	stream_printf(GDKout,"Succeeded to close all session\n");
#endif
	return MAL_SUCCEED;
}

str
SERVERdisconnectWithAlias(int *key, str *dbalias){
	int i;

	mal_set_lock(mal_contextLock,"SERVERdisconnectWithAlias");

	for(i=0; i<MAXSESSIONS; i++)
		 if( SERVERsessions[i].dbalias &&
			 strcmp(SERVERsessions[i].dbalias, *dbalias)==0){
				SERVERsessions[i].c = NULL;
				*key = SERVERsessions[i].key;
				mapi_disconnect(SERVERsessions[i].mid);
				break;
		}

	if( i==MAXSESSIONS){
		mal_unset_lock(mal_contextLock,"SERVERdisconnectWithAlias");
		throw(IO, "mserver.disconnect", "Impossible to close session for db_alias: '%s'", dbalias);
	}

	mal_unset_lock(mal_contextLock,"SERVERdisconnectWithAlias");

#ifdef MAPI_TEST
	stream_printf(GDKout,"Succeeded to close all session\n");
#endif

	return MAL_SUCCEED;
}

str
SERVERconnect(int *key, str *host, int *port, str *username, str *password, str *lang){
	return SERVERconnectAll(key,host,port,username,password,lang, FALSE);
}
str
SERVERconnectssl(int *key, str *host, int *port, str *username, str *password, str *lang){
	return SERVERconnectAll(key,host,port,username,password,lang, TRUE);
}


str
SERVERreconnectAlias(int *key, str *host, int *port, str  * dbalias, str *username, str *password, str *lang){
	int i;
	str msg=MAL_SUCCEED;

	for(i=0; i<MAXSESSIONS; i++)
	 if( SERVERsessions[i].key && 
		 SERVERsessions[i].dbalias &&
		 strcmp(SERVERsessions[i].dbalias, *dbalias)==0){
			*key = SERVERsessions[i].key;
			return msg;
	}

	msg= SERVERconnectAll(key, host, port, username, password, lang, FALSE);
	if( msg == MAL_SUCCEED)
		msg = SERVERsetAlias(&i, key, dbalias);
	return msg;
}
str
SERVERreconnectWithoutAlias(int *key,str *host, int *port, str *username, str *password, str *lang){
	int i;
	str msg=MAL_SUCCEED, nme= "anonymous";

	for(i=0; i<MAXSESSIONS; i++)
	 if( SERVERsessions[i].key ){
			*key = SERVERsessions[i].key;
			return msg;
	}

	msg= SERVERconnectAll(key, host, port, username, password, lang, FALSE);
	if( msg == MAL_SUCCEED)
		msg = SERVERsetAlias(&i, key, &nme);
	return msg;
}
@-
@= accessTest
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== @1 ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.@2","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */
@c
str
SERVERsetAlias(int *ret, int *key, str *dbalias){
	int i;
	Mapi mid;
	@:accessTest(*key,setAlias)@
    SERVERsessions[i].dbalias= GDKstrdup(*dbalias);
	*ret = 0;
	return MAL_SUCCEED;
}
str
SERVERlookup(int *ret, str *dbalias)
{
	int i;
	for(i=0; i< MAXSESSIONS; i++)
	if( SERVERsessions[i].dbalias &&
		strcmp(SERVERsessions[i].dbalias, *dbalias)==0){
		*ret= SERVERsessions[i].key; 
		return MAL_SUCCEED;
	}
	throw(MAL, "mserver.lookup", "Could not find database connection");
}
str
SERVERtrace(int *ret, int *key, int *flag){
	(void )ret;
	mapi_trace(SERVERsessions[*key].mid,*flag);
	return MAL_SUCCEED;
}
str
SERVERdisconnect(int *ret, int *key){
	int i;
	Mapi mid;
	@:accessTest(*key,disconnect)@
	mapi_disconnect(mid);
	if( SERVERsessions[i].dbalias)
		GDKfree(SERVERsessions[i].dbalias);
	SERVERsessions[i].dbalias= 0;
	*ret = 0;
	return MAL_SUCCEED;
}
str
SERVERdestroy(int *ret, int *key){
	int i;
	Mapi mid;
	@:accessTest(*key,destroy)@
	mapi_destroy(mid);
	SERVERsessions[i].c= 0;
	if( SERVERsessions[i].dbalias)
		GDKfree(SERVERsessions[i].dbalias);
	SERVERsessions[i].dbalias= 0;
	*ret = 0;
	return MAL_SUCCEED;
}

str
SERVERreconnect(int *ret, int *key){
	int i;
	Mapi mid;
	@:accessTest(*key,destroy)@
	mapi_reconnect(mid);
	*ret = 0;
	return MAL_SUCCEED;
}
str
SERVERping(int *ret, int *key){
	int i;
	Mapi mid;
	@:accessTest(*key,destroy)@
	*ret= mapi_ping(mid);
	return MAL_SUCCEED;
}
str 
SERVERquery(int *ret, int *key, str *qry){
	Mapi mid;
	MapiHdl hdl=0;
	int i;
	@:accessTest(*key,query)@
	if( SERVERsessions[i].hdl)
		mapi_close_handle(SERVERsessions[i].hdl);
	SERVERsessions[i].hdl = mapi_query(mid, *qry);
	@:catchErrors(mserver.query)@
	*ret = *key;
	return MAL_SUCCEED;
}
str 
SERVERquery_handle(int *ret, int *key, str *qry){
	Mapi mid;
	MapiHdl hdl=0;
	int i;
	@:accessTest(*key,query_handle)@
	mapi_query_handle(SERVERsessions[i].hdl, *qry);
	@:catchErrors(mserver.query_handle)@
	*ret = *key;
	return MAL_SUCCEED;
}
str
SERVERquery_array(MalBlkPtr mb, MalStkPtr stk, InstrPtr pc){
	(void) mb; (void) stk; (void) pc;
	throw(MAL, "mserver.query_array","not yet implemented");
}
str 
SERVERprepare(int *ret, int *key, str *qry){
	Mapi mid;
	int i;
	@:accessTest(*key,prepare)@
	if( SERVERsessions[i].hdl)
		mapi_close_handle(SERVERsessions[i].hdl);
	SERVERsessions[i].hdl= mapi_prepare(mid, *qry);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.prepare",
			mapi_result_error(SERVERsessions[i].hdl));
	*ret = *key;
	return MAL_SUCCEED;
}
str 
SERVERexecute(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,execute)@
	mapi_execute(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.execute",
			mapi_result_error(SERVERsessions[i].hdl));
	*ret = *key;
	return MAL_SUCCEED;
}
str 
SERVERfinish(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,finish)@
	mapi_finish(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.finish",
			mapi_result_error(SERVERsessions[i].hdl));
	*ret = *key;
	return MAL_SUCCEED;
}
str 
SERVERget_row_count(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,get_row_count)@
	*ret= mapi_get_row_count(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.get_row_count",
			mapi_result_error(SERVERsessions[i].hdl));
	return MAL_SUCCEED;
}
str 
SERVERget_field_count(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,get_field_count)@
	*ret= mapi_get_field_count(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.get_field_count",