fixes #872 create unified nng_stream API

This is a major change, and includes changes to use a polymorphic
stream API for all transports.  There have been related bugs fixed
along the way.  Additionally the man pages have changed.

The old non-polymorphic APIs are removed now.  This is a breaking
change, but the old APIs were never part of any released public API.

1 files changed, 80 insertions, 69 deletions

diff --git a/src/platform/windows/win_ipcdial.c b/src/platform/windows/win_ipcdial.c
index 98d848ae..be2a82b3 100644
--- a/src/platform/windows/win_ipcdial.c
+++ b/src/platform/windows/win_ipcdial.c
@@ -1,7 +1,7 @@
 //
-// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2018 Capitar IT Group BV <info@capitar.com>
-// Copyright 2018 Devolutions <info@devolutions.net>
+// Copyright 2019 Devolutions <info@devolutions.net>
 //
 // This software is supplied under the terms of the MIT License, a
 // copy of which should be located in the distribution where this
@@ -15,19 +15,14 @@
 
 #include <stdio.h>
 
-int
-nni_ipc_dialer_init(nni_ipc_dialer **dp)
-{
-	nni_ipc_dialer *d;
-
-	if ((d = NNI_ALLOC_STRUCT(d)) == NULL) {
-		return (NNG_ENOMEM);
-	}
-	d->closed = false;
-	nni_aio_list_init(&d->aios);
-	*dp = d;
-	return (0);
-}
+typedef struct ipc_dialer {
+	nng_stream_dialer sd;
+	bool              closed; // dialers are locked by the worker lock
+	nni_list          aios;
+	nni_list_node     node; // node on worker list
+	char *            path;
+	nni_sockaddr      sa;
+} ipc_dialer;
 
 // Windows IPC is a bit different on the client side.  There is no
 // support for asynchronous connection, but we can fake it with a
@@ -52,7 +47,7 @@ ipc_dial_thr(void *arg)
 
 	nni_mtx_lock(&w->mtx);
 	for (;;) {
-		nni_ipc_dialer *d;
+		ipc_dialer *d;
 
 		if (w->exit) {
 			break;
@@ -63,21 +58,19 @@ ipc_dial_thr(void *arg)
 		}
 
 		while ((d = nni_list_first(&w->workers)) != NULL) {
-			nni_ipc_conn *c;
-			nni_aio *     aio;
-			HANDLE        f;
-			int           rv;
-			char *        path;
+			nng_stream *c;
+			nni_aio *   aio;
+			HANDLE      f;
+			int         rv;
 
 			if ((aio = nni_list_first(&d->aios)) == NULL) {
 				nni_list_remove(&w->workers, d);
 				continue;
 			}
 
-			path = nni_aio_get_prov_extra(aio, 0);
-
-			f = CreateFileA(path, GENERIC_READ | GENERIC_WRITE, 0,
-			    NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
+			f = CreateFileA(d->path, GENERIC_READ | GENERIC_WRITE,
+			    0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED,
+			    NULL);
 
 			if (f == INVALID_HANDLE_VALUE) {
 				switch ((rv = GetLastError())) {
@@ -99,29 +92,20 @@ ipc_dial_thr(void *arg)
 					break;
 				}
 				nni_list_remove(&d->aios, aio);
-				nni_aio_set_prov_extra(aio, 0, NULL);
-				nni_strfree(path);
 				nni_aio_finish_error(aio, rv);
 				continue;
 			}
 
 			nni_list_remove(&d->aios, aio);
-			nni_aio_set_prov_extra(aio, 0, NULL);
 
 			if (((rv = nni_win_io_register(f)) != 0) ||
-			    ((rv = nni_win_ipc_conn_init(&c, f)) != 0)) {
+			    ((rv = nni_win_ipc_init(&c, f, &d->sa, true)) !=
+			        0)) {
 				DisconnectNamedPipe(f);
 				CloseHandle(f);
 				nni_aio_finish_error(aio, rv);
-				nni_strfree(path);
 				continue;
 			}
-			c->dialer             = true;
-			c->sa.s_ipc.sa_family = NNG_AF_IPC;
-			snprintf(c->sa.s_ipc.sa_path,
-			    sizeof(c->sa.s_ipc.sa_path), "%s",
-			    path + strlen(IPC_PIPE_PREFIX));
-			nni_strfree(path);
 			nni_aio_set_output(aio, 0, c);
 			nni_aio_finish(aio, 0, 0);
 		}
@@ -140,27 +124,23 @@ ipc_dial_thr(void *arg)
 static void
 ipc_dial_cancel(nni_aio *aio, void *arg, int rv)
 {
-	nni_ipc_dialer *d = arg;
-	ipc_dial_work * w = &ipc_connecter;
+	ipc_dialer *   d = arg;
+	ipc_dial_work *w = &ipc_connecter;
 
 	nni_mtx_lock(&w->mtx);
 	if (nni_aio_list_active(aio)) {
-		char *path;
 		if (nni_list_active(&w->waiters, d)) {
 			nni_list_remove(&w->waiters, d);
 			nni_cv_wake(&w->cv);
 		}
 		nni_aio_list_remove(aio);
-		path = nni_aio_get_prov_extra(aio, 0);
-		nni_aio_set_prov_extra(aio, 0, NULL);
-		nni_strfree(path);
 		nni_aio_finish_error(aio, rv);
 	}
 	nni_mtx_unlock(&w->mtx);
 }
 
-void
-nni_ipc_dialer_dial(nni_ipc_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
+static void
+ipc_dialer_dial(ipc_dialer *d, nni_aio *aio)
 {
 	ipc_dial_work *w = &ipc_connecter;
 	char *         path;
@@ -169,12 +149,8 @@ nni_ipc_dialer_dial(nni_ipc_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
-	if (sa->s_family != NNG_AF_IPC) {
-		nni_aio_finish_error(aio, NNG_EADDRINVAL);
-		return;
-	}
 	if ((rv = nni_asprintf(
-	         &path, IPC_PIPE_PREFIX "%s", sa->s_ipc.sa_path)) != 0) {
+	         &path, IPC_PIPE_PREFIX "%s", d->sa.s_ipc.sa_path)) != 0) {
 		nni_aio_finish_error(aio, rv);
 		return;
 	}
@@ -182,19 +158,16 @@ nni_ipc_dialer_dial(nni_ipc_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 	nni_mtx_lock(&w->mtx);
 	if ((rv = nni_aio_schedule(aio, ipc_dial_cancel, d)) != 0) {
 		nni_mtx_unlock(&w->mtx);
-		nni_strfree(path);
 		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
 	if (d->closed) {
 		nni_mtx_unlock(&w->mtx);
-		nni_strfree(path);
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 		return;
 	}
 
-	nni_aio_set_prov_extra(aio, 0, path);
 	nni_list_append(&d->aios, aio);
 	if (nni_list_first(&d->aios) == aio) {
 		nni_list_append(&w->waiters, d);
@@ -203,16 +176,10 @@ nni_ipc_dialer_dial(nni_ipc_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 	nni_mtx_unlock(&w->mtx);
 }
 
-void
-nni_ipc_dialer_fini(nni_ipc_dialer *d)
-{
-	nni_ipc_dialer_close(d);
-	NNI_FREE_STRUCT(d);
-}
-
-void
-nni_ipc_dialer_close(nni_ipc_dialer *d)
+static void
+ipc_dialer_close(void *arg)
 {
+	ipc_dialer *   d = arg;
 	ipc_dial_work *w = &ipc_connecter;
 	nni_aio *      aio;
 
@@ -228,6 +195,17 @@ nni_ipc_dialer_close(nni_ipc_dialer *d)
 	nni_mtx_unlock(&w->mtx);
 }
 
+static void
+ipc_dialer_free(void *arg)
+{
+	ipc_dialer *d = arg;
+	ipc_dialer_close(d);
+	if (d->path) {
+		nni_strfree(d->path);
+	}
+	NNI_FREE_STRUCT(d);
+}
+
 static const nni_option ipc_dialer_options[] = {
 	{
 	    .o_name = NULL,
@@ -235,17 +213,50 @@ static const nni_option ipc_dialer_options[] = {
 };
 
 int
-nni_ipc_dialer_setopt(nni_ipc_dialer *d, const char *name, const void *buf,
-    size_t sz, nni_type t)
+ipc_dialer_setx(
+    void *arg, const char *nm, const void *buf, size_t sz, nni_type t)
+{
+	ipc_dialer *d = arg;
+	return (nni_setopt(ipc_dialer_options, nm, d, buf, sz, t));
+}
+
+int
+ipc_dialer_getx(void *arg, const char *nm, void *buf, size_t *szp, nni_type t)
 {
-	return (nni_setopt(ipc_dialer_options, name, d, buf, sz, t));
+	ipc_dialer *d = arg;
+	return (nni_getopt(ipc_dialer_options, nm, d, buf, szp, t));
 }
 
 int
-nni_ipc_dialer_getopt(
-    nni_ipc_dialer *d, const char *name, void *buf, size_t *szp, nni_type t)
+nni_ipc_dialer_alloc(nng_stream_dialer **dp, const nng_url *url)
 {
-	return (nni_getopt(ipc_dialer_options, name, d, buf, szp, t));
+	ipc_dialer *d;
+	int         rv;
+
+	if ((strcmp(url->u_scheme, "ipc") != 0) || (url->u_path == NULL) ||
+	    (strlen(url->u_path) == 0)) {
+		return (NNG_EADDRINVAL);
+	}
+	if ((d = NNI_ALLOC_STRUCT(d)) == NULL) {
+		return (NNG_ENOMEM);
+	}
+
+	if ((rv = nni_asprintf(&d->path, IPC_PIPE_PREFIX "%s", url->u_path)) !=
+	    0) {
+		NNI_FREE_STRUCT(d);
+		return (rv);
+	}
+	snprintf(d->sa.s_ipc.sa_path, NNG_MAXADDRLEN, "%s", url->u_path);
+	d->sa.s_ipc.sa_family = NNG_AF_IPC;
+	d->closed             = false;
+	d->sd.sd_free         = ipc_dialer_free;
+	d->sd.sd_close        = ipc_dialer_close;
+	d->sd.sd_dial         = ipc_dialer_dial;
+	d->sd.sd_getx         = ipc_dialer_getx;
+	d->sd.sd_setx         = ipc_dialer_setx;
+	nni_aio_list_init(&d->aios);
+	*dp = (void *) d;
+	return (0);
 }
 
 int
@@ -254,8 +265,8 @@ nni_win_ipc_sysinit(void)
 	int            rv;
 	ipc_dial_work *worker = &ipc_connecter;
 
-	NNI_LIST_INIT(&worker->workers, nni_ipc_dialer, node);
-	NNI_LIST_INIT(&worker->waiters, nni_ipc_dialer, node);
+	NNI_LIST_INIT(&worker->workers, ipc_dialer, node);
+	NNI_LIST_INIT(&worker->waiters, ipc_dialer, node);
 
 	nni_mtx_init(&worker->mtx);
 	nni_cv_init(&worker->cv, &worker->mtx);