1 files changed, 28 insertions, 11 deletions

diff --git a/src/core/pipe.c b/src/core/pipe.c
index 58a31d7f..bf2e64fe 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -16,23 +16,23 @@
 
 // nni_pipe_id returns the 32-bit pipe id, which can be used in backtraces.
 uint32_t
-nni_pipe_id(nni_pipe * p)
+nni_pipe_id(nni_pipe *p)
 {
 	return (p->p_id);
 }
 
 
 int
-nni_pipe_send(nni_pipe * p, nng_msg *msg)
+nni_pipe_send(nni_pipe *p, nng_msg *msg)
 {
-	return (p->p_ops.p_send(p->p_tran, msg));
+	return (p->p_ops.p_send(p->p_data, msg));
 }
 
 
 int
-nni_pipe_recv(nni_pipe * p, nng_msg **msgp)
+nni_pipe_recv(nni_pipe *p, nng_msg **msgp)
 {
-	return (p->p_ops.p_recv(p->p_tran, msgp));
+	return (p->p_ops.p_recv(p->p_data, msgp));
 }
 
 
@@ -40,22 +40,39 @@ nni_pipe_recv(nni_pipe * p, nng_msg **msgp)
 // subsequent attempts receive or send (including any waiting receive) will
 // simply return NNG_ECLOSED.
 void
-nni_pipe_close(nni_pipe * p)
+nni_pipe_close(nni_pipe *p)
 {
-	p->p_ops.p_close(p->p_tran);
+	p->p_ops.p_close(p->p_data);
 }
 
 
 uint16_t
-nni_pipe_peer(nni_pipe * p)
+nni_pipe_peer(nni_pipe *p)
 {
-	return (p->p_ops.p_peer(p->p_tran));
+	return (p->p_ops.p_peer(p->p_data));
 }
 
 
 void
-nni_pipe_destroy(nni_pipe * p)
+nni_pipe_destroy(nni_pipe *p)
 {
-	p->p_ops.p_destroy(p->p_tran);
+	if (p->p_data != NULL) {
+		p->p_ops.p_destroy(p->p_data);
+	}
 	nni_free(p, sizeof (*p));
 }
+
+
+int
+nni_pipe_create(nni_pipe **pp, const nni_pipe_ops *ops)
+{
+	nni_pipe *p;
+
+	if ((p = nni_alloc(sizeof (*p))) == NULL) {
+		return (NNG_ENOMEM);
+	}
+	p->p_data = NULL;
+	p->p_ops = *ops;
+	p->p_id = nni_plat_nextid();
+	return (0);
+}