fixes #522 Separate out the endpoint plumbing

This separates the plumbing for endpoints into distinct
dialer and listeners.  Some of the transports could benefit
from further separation, but we've done some rather larger
separation e.g. for the websocket transport.

IPC would be a good one to update later, when we start looking
at exposing a more natural underlying API.

1 files changed, 38 insertions, 19 deletions

diff --git a/src/core/pipe.c b/src/core/pipe.c
index 93fbae99..a42cdeff 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -26,7 +26,8 @@ struct nni_pipe {
 	nni_list_node      p_sock_node;
 	nni_list_node      p_ep_node;
 	nni_sock *         p_sock;
-	nni_ep *           p_ep;
+	nni_listener *     p_listener;
+	nni_dialer *       p_dialer;
 	bool               p_closed;
 	bool               p_stop;
 	bool               p_cbs;
@@ -98,7 +99,7 @@ nni_pipe_sys_fini(void)
 	}
 }
 
-static void
+void
 nni_pipe_destroy(nni_pipe *p)
 {
 	bool cbs;
@@ -126,9 +127,8 @@ nni_pipe_destroy(nni_pipe *p)
 
 	// We have exclusive access at this point, so we can check if
 	// we are still on any lists.
-	if (nni_list_node_active(&p->p_ep_node)) {
-		nni_ep_pipe_remove(p->p_ep, p);
-	}
+	nni_dialer_remove_pipe(p->p_dialer, p);     // dialer may be NULL
+	nni_listener_remove_pipe(p->p_listener, p); // listener may be NULL
 
 	if (nni_list_node_active(&p->p_sock_node)) {
 		nni_sock_pipe_remove(p->p_sock, p);
@@ -303,12 +303,10 @@ nni_pipe_start_cb(void *arg)
 }
 
 int
-nni_pipe_create(nni_ep *ep, void *tdata)
+nni_pipe_create2(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata)
 {
 	nni_pipe *          p;
 	int                 rv;
-	nni_tran *          tran  = nni_ep_tran(ep);
-	nni_sock *          sock  = nni_ep_sock(ep);
 	void *              sdata = nni_sock_proto_data(sock);
 	nni_proto_pipe_ops *pops  = nni_sock_proto_pipe_ops(sock);
 
@@ -324,7 +322,6 @@ nni_pipe_create(nni_ep *ep, void *tdata)
 	p->p_tran_data  = tdata;
 	p->p_proto_ops  = *pops;
 	p->p_proto_data = NULL;
-	p->p_ep         = ep;
 	p->p_sock       = sock;
 	p->p_closed     = false;
 	p->p_stop       = false;
@@ -348,16 +345,27 @@ nni_pipe_create(nni_ep *ep, void *tdata)
 	}
 
 	if ((rv != 0) ||
-	    ((rv = pops->pipe_init(&p->p_proto_data, p, sdata)) != 0) ||
-	    ((rv = nni_ep_pipe_add(ep, p)) != 0) ||
-	    ((rv = nni_sock_pipe_add(sock, p)) != 0)) {
+	    ((rv = pops->pipe_init(&p->p_proto_data, p, sdata)) != 0)) {
 		nni_pipe_destroy(p);
 		return (rv);
 	}
 
+	*pp = p;
 	return (0);
 }
 
+void
+nni_pipe_set_listener(nni_pipe *p, nni_listener *l)
+{
+	p->p_listener = l;
+}
+
+void
+nni_pipe_set_dialer(nni_pipe *p, nni_dialer *d)
+{
+	p->p_dialer = d;
+}
+
 int
 nni_pipe_getopt(
     nni_pipe *p, const char *name, void *val, size_t *szp, nni_opt_type t)
@@ -371,7 +379,13 @@ nni_pipe_getopt(
 		return (o->o_get(p->p_tran_data, val, szp, t));
 	}
 	// Maybe the endpoint knows?
-	return (nni_ep_getopt(p->p_ep, name, val, szp, t));
+	if (p->p_dialer != NULL) {
+		return (nni_dialer_getopt(p->p_dialer, name, val, szp, t));
+	}
+	if (p->p_listener != NULL) {
+		return (nni_listener_getopt(p->p_listener, name, val, szp, t));
+	}
+	return (NNG_ENOTSUP);
 }
 
 void
@@ -409,15 +423,20 @@ nni_pipe_sock_id(nni_pipe *p)
 }
 
 uint32_t
-nni_pipe_ep_id(nni_pipe *p)
+nni_pipe_listener_id(nni_pipe *p)
 {
-	return (nni_ep_id(p->p_ep));
+	if (p->p_listener != NULL) {
+		return (nni_listener_id(p->p_listener));
+	}
+	return (0);
 }
-
-int
-nni_pipe_ep_mode(nni_pipe *p)
+uint32_t
+nni_pipe_dialer_id(nni_pipe *p)
 {
-	return (nni_ep_mode(p->p_ep));
+	if (p->p_dialer != NULL) {
+		return (nni_dialer_id(p->p_dialer));
+	}
+	return (0);
 }
 
 static void