- start working on separating out stream clients and dgram clients

3 files changed, 138 insertions, 1 deletions

diff --git a/src/plugin.c b/src/plugin.c
index e8b0e70..e291222 100644
--- a/src/plugin.c
+++ b/src/plugin.c
@@ -41,6 +41,7 @@ struct plugin_state {
 	PLArenaPool *arena;
 	pthread_t tid;
 	Slapi_ComponentId *plugin_identity;
+	Slapi_PluginDesc *plugin_desc;
 	int resvport;
 	int n_listeners;
 	int listenfd[4];
@@ -772,8 +773,9 @@ nis_plugin_init(Slapi_PBlock *pb)
 	slapi_pblock_set(pb, SLAPI_PLUGIN_START_FN, &plugin_start);
 	slapi_pblock_set(pb, SLAPI_PLUGIN_CLOSE_FN, &plugin_close);
 	slapi_pblock_get(pb, SLAPI_PLUGIN_IDENTITY, &state->plugin_identity);
+	state->plugin_desc = &plugin_description;
 	slapi_pblock_set(pb, SLAPI_PLUGIN_PRIVATE, state);
-	slapi_log_error(SLAPI_LOG_PLUGIN, plugin_description.spd_id,
+	slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc.spd_id,
 			"registered plugin hooks\n");
 	return 0;
 }
diff --git a/src/stream.c b/src/stream.c
new file mode 100644
index 0000000..e58f54a
--- /dev/null
+++ b/src/stream.c
@@ -0,0 +1,129 @@
+#ifdef HAVE_CONFIG_H
+#include "../config.h"
+#endif
+
+#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 "stream.h"
+
+struct stream_client_thread_parms {
+	int client;
+	struct plugin_state *state;
+};
+
+static void *
+stream_client_thread(void *arg)
+{
+	int i, fraglen, last, ret, count;
+	int32_t len, nlen;
+	char buf[65536];
+	struct pollfd pollfd;
+
+	struct stream_client_thread_parms *parms = arg;
+	struct plugin_state *state = parms->state;
+	int client = parms->client;
+	free(parms);
+
+	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,
+				"error setting new client connection to be "
+				"non-blocking\n");
+		close(client);
+		client = -1;
+		return NULL;
+	}
+	while (client != -1) {
+		count = 0;
+		pollfd.fd = client;
+		pollfd.events = POLLIN | POLLHUP;
+		if (poll(&pollfd, 1, 60000) > 0) {
+			i = read(client, buf + count, sizeof(buf) - count);
+			switch (i) {
+			case -1:
+				if (errno == EAGAIN) {
+					continue;
+				}
+				slapi_log_error(SLAPI_LOG_PLUGIN,
+						state->plugin_desc.spd_id,
+						"communication error\n");
+				close(client);
+				client = -1;
+				break;
+			case 0:
+				slapi_log_error(SLAPI_LOG_PLUGIN,
+						state->plugin_desc.spd_id,
+						"client closed connection\n");
+				close(client);
+				client = -1;
+				break;
+			default:
+				count += i;
+			}
+			if (count > 4) {
+				memcpy(&nlen, buf, 4);
+				len = ntohl(nlen);
+				last = ((len >> 31) == 1);
+				len &= 0x7fffffff;
+				if (count >= len + 4) {
+					/* we have a whole request */
+					process_request(state, client,
+						        buf + 4, len, 1);
+					memmove(buf, buf + len + 4,
+						count - (len + 4));
+					count -= (len + 4);
+				}
+			}
+		} else {
+			slapi_log_error(SLAPI_LOG_PLUGIN,
+					state->plugin_desc.spd_id,
+					"connection timeout\n");
+			close(client);
+			client = -1;
+		}
+	}
+	close(client);
+	return NULL;
+}
+
+void
+stream_client_start(struct plugin_state *state, int client)
+{
+	pthread_t thread;
+	struct stream_client_thread_parms *parms;
+
+	parms = malloc(sizeof(*parms));
+	if (parms == NULL) {
+		slapi_log_error(SLAPI_LOG_PLUGIN,
+				state->plugin_desc.spd_id,
+				"out of memory\n");
+		close(client);
+		return;
+	}
+	parms->client = client;
+	parms->state = state;
+
+	if (pthread_create(&thread, NULL, &stream_client_thread, parms) != 0) {
+		slapi_log_error(SLAPI_LOG_PLUGIN,
+				state->plugin_desc.spd_id,
+				"error starting thread\n");
+		close(client);
+	}
+	return;
+}
diff --git a/src/stream.h b/src/stream.h
new file mode 100644
index 0000000..443e3cd
--- /dev/null
+++ b/src/stream.h
@@ -0,0 +1,6 @@
+#ifndef stream_h
+#define stream_h
+
+void stream_client_start(struct plugin_state *state, int client);
+
+#endif