--- libaitrpc/src/cli.c	2013/01/17 17:01:51	1.13.4.3
+++ libaitrpc/src/cli.c	2013/03/07 23:03:03	1.13.4.4
@@ -3,7 +3,7 @@
 *  by Michael Pounov <misho@openbsd-bg.org>
 *
 * $Author: misho $
-* $Id: cli.c,v 1.13.4.3 2013/01/17 17:01:51 misho Exp $
+* $Id: cli.c,v 1.13.4.4 2013/03/07 23:03:03 misho Exp $
 *
 **************************************************************************
 The ELWIX and AITNET software is distributed under the following
@@ -263,67 +263,30 @@ rpc_cli_closeClient(rpc_cli_t ** __restrict cli)
 	*cli = NULL;
 }
 
-
 /*
- * rpc_cli_execCall() - Execute RPC call
+ * rpc_pkt_Send() - Send RPC packet
  *
- * @cli = RPC Client session
- * @noreply = We not want RPC reply
- * @tag = Function tag for execution
- * @in_vars = IN RPC call array of rpc values, may be NULL
- * @out_vars = OUT returned array of rpc values, if !=NULL must be free after use with ait_freeVars()
- * return: -1 error or != -1 ok result
+ * @sock = Socket
+ * @type = Type of socket
+ * @sa = Server address
+ * @pkt = RPC packet
+ * @len = Length of packet
+ * return: -1 error or !=-1 sended bytes
  */
 int
-rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short tag, 
-		array_t * __restrict in_vars, array_t ** __restrict out_vars)
+rpc_pkt_Send(int sock, int type, sockaddr_t * __restrict sa, ait_val_t * __restrict pkt, int len)
 {
-	struct tagRPCCall *rpc;
-	int ret = 0, wlen = sizeof(struct tagRPCCall);
-	uint16_t crc;
-	u_char *buf;
 	struct pollfd pfd;
-	sockaddr_t sa;
-	socklen_t salen;
+	int ret;
+	u_char *buf;
 
-	if (!cli) {
-		rpc_SetErr(EINVAL, "Can`t execute call because parameter is null or invalid!");
+	if (!pkt) {
+		rpc_SetErr(EINVAL, "Invalid argument(s)!");
 		return -1;
 	} else
-		buf = AIT_GET_BUF(&cli->cli_buf);
-	if (out_vars)
-		*out_vars = NULL;
+		buf = AIT_GET_BUF(pkt);
 
-	memset(&sa, 0, sizeof sa);
-
-	/* prepare RPC call */
-	rpc = (struct tagRPCCall*) buf;
-	rpc_addPktSession(&rpc->call_session, cli->cli_parent);
-	rpc->call_argc = htons(array_Size(in_vars));
-	rpc->call_tag = htons(tag);
-	rpc->call_seq = htons(random() % USHRT_MAX);
-
-	/* set reply */
-	rpc->call_req.flags = noreply ? RPC_NOREPLY : RPC_REPLY;
-
-	if (array_Size(in_vars)) {
-		/* marshaling variables */
-		ret = ait_vars2buffer(buf + wlen, AIT_LEN(&cli->cli_buf) - wlen, in_vars);
-		if (ret == -1) {
-			rpc_SetErr(EBADRPC, "Failed to prepare RPC packet values");
-			return -1;
-		} else
-			wlen += ret;
-	}
-
-	/* total packet length */
-	rpc->call_len = htons(wlen);
-
-	/* calculate CRC */
-	rpc->call_crc ^= rpc->call_crc;
-	rpc->call_crc = htons(crcFletcher16((u_short*) buf, wlen / 2));
-
-	pfd.fd = cli->cli_sock;
+	pfd.fd = sock;
 	pfd.events = POLLOUT;
 	if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
 			pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
@@ -334,28 +297,55 @@ rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short 
 		return -1;
 	}
 	do {
-		if (cli->cli_id == SOCK_STREAM)
-			ret = send(cli->cli_sock, buf, wlen, MSG_NOSIGNAL);
-		else
-			ret = sendto(cli->cli_sock, buf, wlen, MSG_NOSIGNAL, 
-					&cli->cli_sa.sa, cli->cli_sa.sa.sa_len);
+		if (type == SOCK_STREAM)
+			ret = send(sock, buf, len, MSG_NOSIGNAL);
+		else if (sa)
+			ret = sendto(sock, buf, len, MSG_NOSIGNAL, &sa->sa, sa->sa.sa_len);
+		else {
+			rpc_SetErr(EINVAL, "Invalid argument(s)!");
+			return -1;
+		}
 		if (ret == -1) {
 			if (errno == EAGAIN)
 				continue;
 			LOGERR;
 			return -1;
-		} else if (ret != wlen) {
+		} else if (ret != len) {
 			rpc_SetErr(EPROCUNAVAIL, "RPC request, should be send %d bytes, "
-					"really sended %d bytes", wlen, ret);
+					"really sended %d bytes", len, ret);
 			return -1;
 		}
 	} while (0);
 
-	if (noreply)	/* we not want reply */
-		return 0;
+	return ret;
+}
 
-	wlen = 0;
+/*
+ * rpc_pkt_Receive() - Receive RPC packet
+ *
+ * @sock = Socket
+ * @type = Type of socket
+ * @sa = Server address
+ * @pkt = RPC packet
+ * return: -1 error or !=-1 sended bytes
+ */
+int
+rpc_pkt_Receive(int sock, int type, sockaddr_t * __restrict sa, ait_val_t * __restrict pkt)
+{
+	struct pollfd pfd;
+	int ret, len = 0;
+	u_char *buf;
+	sockaddr_t sa2;
+	socklen_t salen;
+
+	if (!pkt) {
+		rpc_SetErr(EINVAL, "Invalid argument(s)!");
+		return -1;
+	} else
+		buf = AIT_GET_BUF(pkt);
+
 	/* reply from RPC server */
+	pfd.fd = sock;
 	pfd.events = POLLIN | POLLPRI;
 	do {
 		if ((ret = poll(&pfd, 1, DEF_RPC_TIMEOUT * 1000)) < 1 || 
@@ -363,22 +353,23 @@ rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short 
 			if (ret) {
 				LOGERR;
 			} else {
-				if (wlen++ < 7)
+				if (len++ < 7)
 					continue;
 				else
-					rpc_SetErr(ETIMEDOUT, "Timeout, no answer from RPC server");
+					rpc_SetErr(ETIMEDOUT, 
+							"Timeout, no answer from RPC server");
 			}
 
 			return -1;
 		}
 
-		memset(buf, 0, AIT_LEN(&cli->cli_buf));
-		if (cli->cli_id == SOCK_STREAM)
-			ret = recv(cli->cli_sock, buf, AIT_LEN(&cli->cli_buf), 0);
+		memset(buf, 0, AIT_LEN(pkt));
+		if (type == SOCK_STREAM)
+			ret = recv(sock, buf, AIT_LEN(pkt), 0);
 		else {
-			salen = sa.ss.ss_len = sizeof(sockaddr_t);
-			ret = recvfrom(cli->cli_sock, buf, AIT_LEN(&cli->cli_buf), 0, 
-					&sa.sa, &salen);
+			memset(&sa2, 0, sizeof sa2);
+			salen = sa2.ss.ss_len = sizeof(sockaddr_t);
+			ret = recvfrom(sock, buf, AIT_LEN(pkt), 0, &sa2.sa, &salen);
 		}
 		if (ret < 1) {
 			if (ret) {
@@ -390,20 +381,105 @@ rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short 
 			return -1;
 		}
 
-		if (cli->cli_id == SOCK_DGRAM) {
-			/* check for response from known address */
-			if (e_addrcmp(&cli->cli_sa, &sa, 42)) {
-				rpc_SetErr(ERPCMISMATCH, "Received RPC response from unknown address");
+		/* check for response from known address */
+		if (type == SOCK_DGRAM)
+			if (e_addrcmp(sa, &sa2, 42)) {
+				rpc_SetErr(ERPCMISMATCH, 
+						"Received RPC response from unknown address");
 				continue;
 			}
-		}
 	} while (0);
 	if (ret < sizeof(struct tagRPCCall)) {
 		rpc_SetErr(ERPCMISMATCH, "Short RPC packet %d bytes", ret);
 		return -1;
 	}
 
+	return ret;
+}
+
+/*
+ * rpc_pkt_Request() - Build RPC Request packet
+ *
+ * @pkt = Packet buffer
+ * @sess = RPC session info
+ * @tag = Function tag for execution
+ * @vars = Function argument array of values, may be NULL
+ * @noreply = We not want RPC reply
+ * return: -1 error or != -1 prepared bytes into packet
+ */
+int
+rpc_pkt_Request(ait_val_t * __restrict pkt, rpc_sess_t * __restrict sess, u_short tag, 
+		array_t * __restrict vars, int noreply)
+{
+	struct tagRPCCall *rpc;
+	int ret = 0, len = sizeof(struct tagRPCCall);
+	u_char *buf;
+
+	if (!pkt || !sess) {
+		rpc_SetErr(EINVAL, "Invalid argument(s)!");
+		return -1;
+	} else
+		buf = AIT_GET_BUF(pkt);
+
+	/* prepare RPC call */
+	rpc = (struct tagRPCCall*) buf;
+	rpc_addPktSession(&rpc->call_session, sess);
+	rpc->call_tag = htons(tag);
+	rpc->call_seq = htons(random() % USHRT_MAX);
+	if (!vars)
+		rpc->call_argc = 0;
+	else
+		rpc->call_argc = htons(array_Size(vars));
+
+	/* set reply */
+	rpc->call_req.flags = noreply ? RPC_NOREPLY : RPC_REPLY;
+
+	if (array_Size(vars)) {
+		/* marshaling variables */
+		ret = ait_vars2buffer(buf + len, AIT_LEN(pkt) - len, vars);
+		if (ret == -1) {
+			rpc_SetErr(EBADRPC, "Failed to prepare RPC packet values");
+			return -1;
+		} else
+			len += ret;
+	}
+
+	/* total packet length */
+	rpc->call_len = htons(len);
+
 	/* calculate CRC */
+	rpc->call_crc ^= rpc->call_crc;
+	rpc->call_crc = htons(crcFletcher16((u_short*) buf, len / 2));
+
+	return len;
+}
+
+/*
+ * rpc_pkt_Replay() - Decode RPC Replay packet
+ *
+ * @pkt = Packet buffer
+ * @sess = RPC session info
+ * @tag = Function tag
+ * @vars = Function argument array of values, may be NULL
+ * return: -1 error or != -1 return value from function
+ */
+int
+rpc_pkt_Replay(ait_val_t * __restrict pkt, rpc_sess_t * __restrict sess, u_short tag, 
+		array_t ** __restrict vars)
+{
+	struct tagRPCCall *rpc;
+	int len;
+	u_char *buf;
+	uint16_t crc;
+
+	if (!pkt || !sess) {
+		rpc_SetErr(EINVAL, "Invalid argument(s)!");
+		return -1;
+	} else
+		buf = AIT_GET_BUF(pkt);
+
+	rpc = (struct tagRPCCall*) buf;
+	/* calculate CRC */
 	crc = ntohs(rpc->call_crc);
 	rpc->call_crc ^= rpc->call_crc;
 	if (crc != crcFletcher16((u_short*) buf, ntohs(rpc->call_len) / 2)) {
@@ -412,11 +488,10 @@ rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short 
 	}
 
 	/* check RPC packet session info */
-	if (rpc_chkPktSession(&rpc->call_session, cli->cli_parent)) {
+	if (rpc_chkPktSession(&rpc->call_session, sess)) {
 		rpc_SetErr(ERPCMISMATCH, "Get invalid RPC session");
 		return -1;
-	} else
-		wlen = sizeof(struct tagRPCCall);
+	}
 	if (ntohs(rpc->call_tag) != tag) {
 		rpc_SetErr(ERPCMISMATCH, "Get wrong RPC reply");
 		return -1;
@@ -427,27 +502,74 @@ rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short 
 				strerror(ntohl(rpc->call_rep.eno)));
 		return -1;
 	}
-	if (ntohs(rpc->call_argc) * sizeof(ait_val_t) > AIT_LEN(&cli->cli_buf) - wlen) {
-		rpc_SetErr(EMSGSIZE, "Reply RPC packet is too long ...");
+	len = ntohs(rpc->call_argc) * sizeof(ait_val_t);
+	if (len > AIT_LEN(pkt) - sizeof(struct tagRPCCall)) {
+		rpc_SetErr(EMSGSIZE, "Reply RPC packet not enough buffer space ...");
 		return -1;
 	}
+	if (len > ntohs(rpc->call_len) - sizeof(struct tagRPCCall)) {
+		rpc_SetErr(EMSGSIZE, "Reply RPC packet is too short ...");
+		return -1;
+	}
 
 	/* RPC is OK! Go de-marshaling variables ... */
-	if (out_vars && ntohs(rpc->call_argc)) {
+	if (vars && ntohs(rpc->call_argc)) {
 #ifdef CLI_RES_ZCOPY
-		*out_vars = ait_buffer2vars(buf + wlen, AIT_LEN(&cli->cli_buf) - wlen, 
+		*vars = ait_buffer2vars(buf + sizeof(struct tagRPCCall), len, 
 				ntohs(rpc->call_argc), 42);
 #else
-		*out_vars = ait_buffer2vars(buf + wlen, AIT_LEN(&cli->cli_buf) - wlen, 
+		*vars = ait_buffer2vars(buf + sizeof(struct tagRPCCall), len, 
 				ntohs(rpc->call_argc), 0);
 #endif
-		if (!*out_vars) {
+		if (!*vars) {
 			rpc_SetErr(elwix_GetErrno(), "%s", elwix_GetError());
 			return -1;
 		}
 	}
 
-	ret = ntohl(rpc->call_rep.ret);
+	return ntohl(rpc->call_rep.ret);
+}
+
+/*
+ * rpc_cli_execCall() - Execute RPC call
+ *
+ * @cli = RPC Client session
+ * @noreply = We not want RPC reply
+ * @tag = Function tag for execution
+ * @in_vars = IN function argument array of values, may be NULL
+ * @out_vars = OUT returned array of rpc values, if !=NULL must be free after use with ait_freeVars()
+ * return: -1 error or != -1 ok result
+ */
+int
+rpc_cli_execCall(rpc_cli_t *cli, int noreply, u_short tag, 
+		array_t * __restrict in_vars, array_t ** __restrict out_vars)
+{
+	int ret = 0, wlen;
+	u_char *buf;
+
+	if (!cli) {
+		rpc_SetErr(EINVAL, "Can`t execute call because parameter is null or invalid!");
+		return -1;
+	} else
+		buf = AIT_GET_BUF(&cli->cli_buf);
+	if (out_vars)
+		*out_vars = NULL;
+
+	if ((wlen = rpc_pkt_Request(&cli->cli_buf, cli->cli_parent, tag, in_vars, noreply)) == -1)
+		return -1;
+
+	if (rpc_pkt_Send(cli->cli_sock, cli->cli_id, &cli->cli_sa, &cli->cli_buf, wlen) == -1)
+		return -1;
+
+	if (noreply)	/* we not want reply */
+		return 0;
+
+	if (rpc_pkt_Receive(cli->cli_sock, cli->cli_id, &cli->cli_sa, &cli->cli_buf) == -1)
+		return -1;
+
+	if ((wlen = rpc_pkt_Replay(&cli->cli_buf, cli->cli_parent, tag, out_vars)) == -1)
+		return -1;
+
 	return ret;
 }