--- libaitrpc/src/srv.c	2012/11/16 08:33:06	1.12.2.1
+++ libaitrpc/src/srv.c	2012/11/16 10:20:55	1.12.2.3
@@ -3,7 +3,7 @@
 *  by Michael Pounov <misho@openbsd-bg.org>
 *
 * $Author: misho $
-* $Id: srv.c,v 1.12.2.1 2012/11/16 08:33:06 misho Exp $
+* $Id: srv.c,v 1.12.2.3 2012/11/16 10:20:55 misho Exp $
 *
 **************************************************************************
 The ELWIX and AITNET software is distributed under the following
@@ -46,16 +46,103 @@ SUCH DAMAGE.
 #include "global.h"
 
 
+/* SOCK_STREAM */
 static void *acceptClients(sched_task_t *);
 static void *closeClient(sched_task_t *);
-static sched_task_func_t cbProto[SOCK_RAW][2] = {
-	{ acceptClients, closeClient }, 
-	{ acceptClients, closeClient }, 
-	{ NULL, NULL }
+static void *rxPacket(sched_task_t *);
+static void *txPacket(sched_task_t *);
+
+/* SOCK_DGRAM */
+static void *connectClients(sched_task_t *);
+static void *disconnectClient(sched_task_t *);
+
+/* SOCK_RAW */
+
+static sched_task_func_t cbProto[SOCK_RAW + 1][4] = {
+	{ acceptClients, closeClient, rxPacket, txPacket },	/* SOCK_STREAM */
+	{ acceptClients, closeClient, rxPacket, txPacket },	/* SOCK_STREAM */
+	{ connectClients, disconnectClient, NULL, NULL }, 
+	{ NULL, NULL, NULL, NULL }				/* SOCK_RAW */
 };
 
 
+static rpc_cli_t *
+_allocClient(rpc_srv_t * __restrict srv)
+{
+	rpc_cli_t *c = NULL;
+	register int i;
+
+	/* check free slots for connect */
+	for (i = 0; i < io_arraySize(srv->srv_clients) && 
+			(c = io_array(srv->srv_clients, i, rpc_cli_t*)); i++);
+	if (c)	/* no more free slots! */
+		return NULL;
+
+	c = io_malloc(sizeof(rpc_cli_t));
+	if (!c) {
+		LOGERR;
+		srv->srv_kill = 1;
+		return NULL;
+	} else {
+		memset(c, 0, sizeof(rpc_cli_t));
+		io_arraySet(srv->srv_clients, i, c);
+		c->cli_id = i;
+		c->cli_parent = srv;
+	}
+
+	/* alloc empty buffer */
+	AIT_SET_BUF2(&c->cli_buf, 0, srv->srv_netbuf);
+
+	return c;
+}
+
+
 static void *
+connectClients(sched_task_t *task)
+{
+	rpc_srv_t *srv = TASK_ARG(task);
+	rpc_cli_t *c = NULL;
+	socklen_t salen = sizeof(io_sockaddr_t);
+
+	c = _allocClient(srv);
+	if (!c)
+		goto end;
+
+	/* accept client */
+	c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
+	if (c->cli_sock == -1) {
+		LOGERR;
+		AIT_FREE_VAL(&c->cli_buf);
+		io_arrayDel(srv->srv_clients, c->cli_id, 42);
+		goto end;
+	} else
+		fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
+
+	schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c, 
+			c->cli_sock, NULL, 0);
+end:
+	schedReadSelf(task);
+	return NULL;
+}
+
+static void *
+disconnectClient(sched_task_t *task)
+{
+	rpc_cli_t *c = TASK_ARG(task);
+	rpc_srv_t *s = c->cli_parent;
+
+	schedCancelby(TASK_ROOT(task), taskMAX, CRITERIA_ARG, TASK_ARG(task), NULL);
+
+	/* free buffer */
+	AIT_FREE_VAL(&c->cli_buf);
+
+	io_arrayDel(s->srv_clients, c->cli_id, 0);
+	if (c)
+		io_free(c);
+	return NULL;
+}
+
+static void *
 closeClient(sched_task_t *task)
 {
 	rpc_cli_t *c = TASK_ARG(task);
@@ -197,8 +284,6 @@ rxPacket(sched_task_t *task)
 
 	if (!off)
 		memset(buf, 0, AIT_LEN(&c->cli_buf));
-	else
-		memmove(buf, buf + off, AIT_LEN(&c->cli_buf) - off);
 	rlen = recv(TASK_FD(task), buf + off, AIT_LEN(&c->cli_buf) - off, 0);
 	if (rlen < 1) {
 		/* close connection */
@@ -258,7 +343,8 @@ rxPacket(sched_task_t *task)
 
 		/* send RPC reply */
 		if (!noreply)
-			schedWrite(TASK_ROOT(task), txPacket, TASK_ARG(task), TASK_FD(task), rpc, len);
+			schedWrite(TASK_ROOT(task), cbProto[s->srv_proto][CB_TXPACKET], 
+					TASK_ARG(task), TASK_FD(task), rpc, len);
 
 		off += len;
 	} while (rlen > 0);
@@ -273,40 +359,24 @@ acceptClients(sched_task_t *task)
 {
 	rpc_srv_t *srv = TASK_ARG(task);
 	rpc_cli_t *c = NULL;
-	register int i;
 	socklen_t salen = sizeof(io_sockaddr_t);
 
-	/* check free slots for connect */
-	for (i = 0; i < io_arraySize(srv->srv_clients) && 
-			(c = io_array(srv->srv_clients, i, rpc_cli_t*)); i++);
-	if (c)	/* no more free slots! */
+	c = _allocClient(srv);
+	if (!c)
 		goto end;
-	c = io_malloc(sizeof(rpc_cli_t));
-	if (!c) {
-		LOGERR;
-		srv->srv_kill = 1;
-		return NULL;
-	} else {
-		memset(c, 0, sizeof(rpc_cli_t));
-		io_arraySet(srv->srv_clients, i, c);
-		c->cli_id = i;
-		c->cli_parent = srv;
-	}
 
-	/* alloc empty buffer */
-	AIT_SET_BUF2(&c->cli_buf, 0, srv->srv_netbuf);
-
 	/* accept client */
 	c->cli_sock = accept(TASK_FD(task), &c->cli_sa.sa, &salen);
 	if (c->cli_sock == -1) {
 		LOGERR;
 		AIT_FREE_VAL(&c->cli_buf);
-		io_arrayDel(srv->srv_clients, i, 42);
+		io_arrayDel(srv->srv_clients, c->cli_id, 42);
 		goto end;
 	} else
 		fcntl(c->cli_sock, F_SETFL, fcntl(c->cli_sock, F_GETFL) | O_NONBLOCK);
 
-	schedRead(TASK_ROOT(task), rxPacket, c, c->cli_sock, NULL, 0);
+	schedRead(TASK_ROOT(task), cbProto[srv->srv_proto][CB_RXPACKET], c, 
+			c->cli_sock, NULL, 0);
 end:
 	schedReadSelf(task);
 	return NULL;