1 files changed, 32 insertions, 10 deletions

diff --git a/src/core/pipe.c b/src/core/pipe.c
index cc92e6fe..0a7bbed1 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -25,14 +25,14 @@ nni_pipe_id(nni_pipe *p)
 int
 nni_pipe_send(nni_pipe *p, nng_msg *msg)
 {
-	return (p->p_ops.p_send(p->p_data, msg));
+	return (p->p_ops.p_send(p->p_trandata, msg));
 }
 
 
 int
 nni_pipe_recv(nni_pipe *p, nng_msg **msgp)
 {
-	return (p->p_ops.p_recv(p->p_data, msgp));
+	return (p->p_ops.p_recv(p->p_trandata, msgp));
 }
 
 
@@ -42,38 +42,60 @@ nni_pipe_recv(nni_pipe *p, nng_msg **msgp)
 void
 nni_pipe_close(nni_pipe *p)
 {
-	p->p_ops.p_close(p->p_data);
+	nni_socket *sock = p->p_sock;
+
+	p->p_ops.p_close(p->p_trandata);
+
+	nni_mutex_enter(&sock->s_mx);
+	if (!p->p_reap) {
+		// schedule deferred reap/close
+		p->p_reap = 1;
+		if (p->p_active) {
+			nni_list_remove(&sock->s_pipes, p);
+			p->p_active = 0;
+		}
+		nni_list_append(&sock->s_reaps, p);
+		nni_cond_broadcast(&sock->s_cv);
+	}
+	nni_mutex_exit(&sock->s_mx);
 }
 
 
 uint16_t
 nni_pipe_peer(nni_pipe *p)
 {
-	return (p->p_ops.p_peer(p->p_data));
+	return (p->p_ops.p_peer(p->p_trandata));
 }
 
 
 void
 nni_pipe_destroy(nni_pipe *p)
 {
-	if (p->p_data != NULL) {
-		p->p_ops.p_destroy(p->p_data);
+	if (p->p_trandata != NULL) {
+		p->p_ops.p_destroy(p->p_trandata);
 	}
 	nni_free(p, sizeof (*p));
 }
 
 
 int
-nni_pipe_create(nni_pipe **pp, const nni_pipe_ops *ops)
+nni_pipe_create(nni_pipe **pp, nni_endpt *ep)
 {
 	nni_pipe *p;
 
 	if ((p = nni_alloc(sizeof (*p))) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	p->p_data = NULL;
-	p->p_ops = *ops;
+	p->p_trandata = NULL;
+	p->p_protdata = NULL;
+	p->p_ops = *ep->ep_ops.ep_pipe_ops;
 	p->p_id = nni_plat_nextid();
+	p->p_ep = ep;
+	p->p_sock = ep->ep_sock;
+	if (ep->ep_dialer != NULL) {
+		ep->ep_pipe = p;
+	}
+	NNI_LIST_NODE_INIT(&p->p_node);
 	*pp = p;
 	return (0);
 }
@@ -86,5 +108,5 @@ nni_pipe_getopt(nni_pipe *p, int opt, void *val, size_t *szp)
 	if (p->p_ops.p_getopt == NULL) {
 		return (NNG_ENOTSUP);
 	}
-	return (p->p_ops.p_getopt(p->p_data, opt, val, szp));
+	return (p->p_ops.p_getopt(p->p_trandata, opt, val, szp));
 }