Begin testing REQ/REP.

8 files changed, 104 insertions, 2 deletions

diff --git a/src/core/protocol.c b/src/core/protocol.c
index 8669275a..072cc301 100644
--- a/src/core/protocol.c
+++ b/src/core/protocol.c
@@ -17,9 +17,13 @@
 // change, as adding new protocols is not something intended to be done
 // outside of the core.
 extern nni_proto nni_pair_proto;
+extern nni_proto nni_rep_proto;
+extern nni_proto nni_req_proto;
 
 static nni_proto *protocols[] = {
 	&nni_pair_proto,
+	&nni_rep_proto,
+	&nni_req_proto,
 	NULL
 };
 
diff --git a/src/core/socket.c b/src/core/socket.c
index cad4b10a..26ca055e 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -361,6 +361,13 @@ nni_sock_proto(nni_sock *sock)
 }
 
 
+uint16_t
+nni_sock_peer(nni_sock *sock)
+{
+	return (sock->s_proto.proto_peer);
+}
+
+
 int
 nni_sock_dial(nni_sock *sock, const char *addr, nni_ep **epp, int flags)
 {
diff --git a/src/nng.c b/src/nng.c
index 3923480d..79069723 100644
--- a/src/nng.c
+++ b/src/nng.c
@@ -57,6 +57,14 @@ nng_protocol(nng_socket *s)
 }
 
 
+uint16_t
+nng_peer(nng_socket *s)
+{
+	NNI_INIT_VOID();
+	return (nni_sock_peer(s));
+}
+
+
 int
 nng_recvmsg(nng_socket *s, nng_msg **msgp, int flags)
 {
diff --git a/src/nng.h b/src/nng.h
index 9388f202..a0995595 100644
--- a/src/nng.h
+++ b/src/nng.h
@@ -59,6 +59,9 @@ NNG_DECL int nng_close(nng_socket *);
 // nng_protocol returns the protocol number of the socket.
 uint16_t nng_protocol(nng_socket *);
 
+// nng_peer returns the protocol number for the socket's peer.
+uint16_t nng_peer(nng_socket *);
+
 // nng_setopt sets an option for a specific socket.
 NNG_DECL int nng_setopt(nng_socket *, int, const void *, size_t);
 
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index 8d51ee0a..60346390 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -312,6 +312,7 @@ again:
 		}
 	}
 	nni_msgq_signal(uwq, &rp->sigclose);
+	nni_msgq_signal(rp->sendq, &rp->sigclose);
 	nni_pipe_close(pipe);
 }
 
@@ -452,7 +453,7 @@ static nni_proto_pipe nni_rep_proto_pipe = {
 	.pipe_recv	= nni_rep_pipe_recv,
 };
 
-nni_proto nni_rep_protocol = {
+nni_proto nni_rep_proto = {
 	.proto_self		= NNG_PROTO_REP,
 	.proto_peer		= NNG_PROTO_REQ,
 	.proto_name		= "rep",
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index c26d66e8..b8401107 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -118,6 +118,7 @@ nni_req_pipe_init(void **rpp, nni_pipe *pipe, void *rsock)
 	rp->pipe = pipe;
 	rp->sigclose = 0;
 	rp->req = rsock;
+	*rpp = rp;
 	return (0);
 }
 
@@ -414,7 +415,7 @@ static nni_proto_pipe nni_req_proto_pipe = {
 	.pipe_recv	= nni_req_pipe_recv,
 };
 
-nni_proto nni_req_protocol = {
+nni_proto nni_req_proto = {
 	.proto_self		= NNG_PROTO_REQ,
 	.proto_peer		= NNG_PROTO_REP,
 	.proto_name		= "req",
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 0a225c7e..a63e8fc2 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -55,4 +55,5 @@ endif ()
 add_nng_test(idhash 5)
 add_nng_test(list 5)
 add_nng_test(platform 5)
+add_nng_test(reqrep 5)
 add_nng_test(sock 5)
diff --git a/tests/reqrep.c b/tests/reqrep.c
new file mode 100644
index 00000000..9c45d2d4
--- /dev/null
+++ b/tests/reqrep.c
@@ -0,0 +1,77 @@
+//
+// Copyright 2016 Garrett D'Amore <garrett@damore.org>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include "convey.h"
+#include "nng.h"
+
+#include <string.h>
+
+Main({
+	int rv;
+	const char *addr = "inproc://test";
+
+	Test("REQ/REP pattern", {
+		Convey("We can create a REQ socket", {
+			nng_socket *req;
+
+			rv = nng_open(&req, NNG_PROTO_REQ);
+			So(rv == 0);
+			So(req != NULL);
+
+			Reset({
+				nng_close(req);
+			})
+
+			Convey("Protocols match", {
+				So(nng_protocol(req) == NNG_PROTO_REQ);
+				So(nng_peer(req) == NNG_PROTO_REP);
+			})
+		})
+
+		Convey("We can create a REP socket", {
+			nng_socket *rep;
+			rv = nng_open(&rep, NNG_PROTO_REP);
+			So(rv == 0);
+			So(rep != NULL);
+
+			Reset({
+				nng_close(rep);
+			})
+
+			Convey("Protocols match", {
+				So(nng_protocol(rep) == NNG_PROTO_REP);
+				So(nng_peer(rep) == NNG_PROTO_REQ);
+			})
+		})
+
+		Convey("We can create a linked REQ/REP pair", {
+			nng_socket *req;
+			nng_socket *rep;
+
+			rv = nng_open(&rep, NNG_PROTO_REP);
+			So(rv == 0);
+			So(rep != NULL);
+
+			rv = nng_open(&req, NNG_PROTO_REQ);
+			So(rv == 0);
+			So(req != NULL);
+
+			Reset({
+				nng_close(rep);
+				nng_close(req);
+			})
+
+			rv = nng_listen(rep, addr, NULL, NNG_FLAG_SYNCH);
+			So(rv == 0);
+
+			rv = nng_dial(req, addr, NULL, NNG_FLAG_SYNCH);
+			So(rv == 0);
+		})
+	})
+})