- split up some more

1 files changed, 58 insertions, 28 deletions

diff --git a/src/stream.c b/src/stream.c
index ebf8c7f..dbb53bf 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -4,21 +4,15 @@
 
 #include <sys/types.h>
 #include <sys/socket.h>
-#include <sys/stat.h>
 #include <errno.h>
 #include <fcntl.h>
 #include <poll.h>
 #include <pthread.h>
-#include <stdlib.h>
-#include <string.h>
-#include <syslog.h>
 #include <unistd.h>
 
-#include <nspr.h>
-#include <secport.h>
-#include <dirsrv/slapi-plugin.h>
-
+#include "nis.h"
 #include "stream.h"
+#define MAX_CLIENT_IDLE (60 * 1000)
 
 struct stream_client_thread_parms {
 	int client;
@@ -35,26 +29,33 @@ stream_client_thread(void *arg)
 
 	struct stream_client_thread_parms *parms = arg;
 	struct plugin_state *state = parms->state;
-	int client = parms->client;
+	int done, client = parms->client;
 	free(parms);
 
+	slapi_log_error(SLAPI_LOG_PLUGIN,
+			state->plugin_desc->spd_id,
+			"opened client connection %d, thread started\n",
+			client);
 	last = fcntl(client, F_GETFD);
 	if ((last == -1) ||
 	    (fcntl(client, F_SETFD, last | O_NONBLOCK) == -1)) {
 		slapi_log_error(SLAPI_LOG_PLUGIN,
-				state->plugin_desc.spd_id,
+				state->plugin_desc->spd_id,
 				"error setting new client connection to be "
 				"non-blocking\n");
 		close(client);
-		client = -1;
 		return NULL;
 	}
+	slapi_log_error(SLAPI_LOG_PLUGIN,
+			state->plugin_desc->spd_id,
+			"set new client connection to be non-blocking\n");
 	frag_len = 0;
 	record_len = 0;
-	while (client != -1) {
+	done = 0;
+	while (!done) {
 		pollfd.fd = client;
 		pollfd.events = POLLIN | POLLHUP;
-		if (poll(&pollfd, 1, 60 * 1000) > 0) {
+		if (poll(&pollfd, 1, MAX_CLIENT_IDLE) > 0) {
 			i = read(client, fragment + frag_len,
 				 sizeof(fragment) - frag_len);
 			switch (i) {
@@ -63,37 +64,57 @@ stream_client_thread(void *arg)
 					continue;
 				}
 				slapi_log_error(SLAPI_LOG_PLUGIN,
-						state->plugin_desc.spd_id,
+						state->plugin_desc->spd_id,
 						"communication error\n");
-				close(client);
-				client = -1;
+				done = 1;
 				break;
 			case 0:
 				slapi_log_error(SLAPI_LOG_PLUGIN,
-						state->plugin_desc.spd_id,
+						state->plugin_desc->spd_id,
 						"client closed connection\n");
-				close(client);
-				client = -1;
+				done = 1;
 				break;
 			default:
+				slapi_log_error(SLAPI_LOG_PLUGIN,
+						state->plugin_desc->spd_id,
+						"read %d bytes\n", i);
 				frag_len += i;
 				break;
 			}
 			if (frag_len > 4) {
 				memcpy(&nlen, fragment, 4);
 				len = ntohl(nlen);
-				last = ((len >> 31) == 1);
+				last = ((len & 0x80000000) != 0);
 				len &= 0x7fffffff;
+				slapi_log_error(SLAPI_LOG_PLUGIN,
+						state->plugin_desc->spd_id,
+						"expecting %d bytes\n", len);
+				slapi_log_error(SLAPI_LOG_PLUGIN,
+						state->plugin_desc->spd_id,
+						"in: frag_len=%d,record=%d\n",
+						frag_len, record_len);
 				if (frag_len >= len + 4) {
+					slapi_log_error(SLAPI_LOG_PLUGIN,
+							state->plugin_desc->spd_id,
+							"got whole fragment "
+							"(%d bytes)\n", len);
 					/* we have a whole fragment */
 					memmove(record + record_len,
 						fragment + 4, len);
 					record_len += len;
 					if (last) {
-						/* we have a whole record */
-						process_request(state, client,
-								record,
+						slapi_log_error(SLAPI_LOG_PLUGIN,
+								state->plugin_desc->spd_id,
+								"got whole "
+								"record (%d "
+								"bytes)\n",
 								record_len);
+						/* we have a whole record */
+						nis_process_request(state,
+								    client,
+								    NULL, 0,
+								    record,
+								    record_len);
 						record_len = 0;
 					}
 					memmove(fragment,
@@ -101,16 +122,22 @@ stream_client_thread(void *arg)
 						frag_len - (len + 4));
 					frag_len -= (len + 4);
 				}
+				slapi_log_error(SLAPI_LOG_PLUGIN,
+						state->plugin_desc->spd_id,
+						"out: frag_len=%d,record=%d\n",
+						frag_len, record_len);
 			}
 		} else {
 			slapi_log_error(SLAPI_LOG_PLUGIN,
-					state->plugin_desc.spd_id,
+					state->plugin_desc->spd_id,
 					"connection timeout\n");
-			close(client);
-			client = -1;
+			done = 1;
 		}
 	}
 	close(client);
+	slapi_log_error(SLAPI_LOG_PLUGIN,
+			state->plugin_desc->spd_id,
+			"closed client connection %d, thread ending\n", client);
 	return NULL;
 }
 
@@ -123,7 +150,7 @@ stream_client_start(struct plugin_state *state, int client)
 	parms = malloc(sizeof(*parms));
 	if (parms == NULL) {
 		slapi_log_error(SLAPI_LOG_PLUGIN,
-				state->plugin_desc.spd_id,
+				state->plugin_desc->spd_id,
 				"out of memory\n");
 		close(client);
 		return;
@@ -133,9 +160,12 @@ stream_client_start(struct plugin_state *state, int client)
 
 	if (pthread_create(&thread, NULL, &stream_client_thread, parms) != 0) {
 		slapi_log_error(SLAPI_LOG_PLUGIN,
-				state->plugin_desc.spd_id,
+				state->plugin_desc->spd_id,
 				"error starting thread\n");
 		close(client);
 	}
+	slapi_log_error(SLAPI_LOG_PLUGIN,
+			state->plugin_desc->spd_id,
+			"started client-specific thread\n");
 	return;
 }