Test coverage improvements for REQ/REP.

This also fixes a possible bug if mixing poll file descriptors and
contexts on the same socket.  Most folks are unlikely to ever run
into this bug.

At this point the REQ/REP coverage is nearly complete (over 95%).

6 files changed, 952 insertions, 510 deletions

diff --git a/src/protocol/reqrep0/CMakeLists.txt b/src/protocol/reqrep0/CMakeLists.txt
index 4e1b7a8f..46eb7abf 100644
--- a/src/protocol/reqrep0/CMakeLists.txt
+++ b/src/protocol/reqrep0/CMakeLists.txt
@@ -23,7 +23,7 @@ nng_sources_if(NNG_PROTO_REP0 rep.c xrep.c)
 nng_headers_if(NNG_PROTO_REP0 nng/protocol/reqrep0/rep.h)
 nng_defines_if(NNG_PROTO_REP0 NNG_HAVE_REP0)
 
-nng_test(reqrep_test)
+nng_test(req_test)
 nng_test(rep_test)
 nng_test(xrep_test)
 nng_test(xreq_test)
diff --git a/src/protocol/reqrep0/rep.c b/src/protocol/reqrep0/rep.c
index 3e1a34a9..3cc1802a 100644
--- a/src/protocol/reqrep0/rep.c
+++ b/src/protocol/reqrep0/rep.c
@@ -169,6 +169,11 @@ rep0_ctx_send(void *arg, nni_aio *aio)
 		// reply for the single request we got.
 		nni_pollable_clear(&s->writable);
 	}
+	if ((rv = nni_aio_schedule(aio, rep0_ctx_cancel_send, ctx)) != 0) {
+		nni_mtx_unlock(&s->lk);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
 
 	if (len == 0) {
 		nni_mtx_unlock(&s->lk);
@@ -202,11 +207,6 @@ rep0_ctx_send(void *arg, nni_aio *aio)
 		return;
 	}
 
-	if ((rv = nni_aio_schedule(aio, rep0_ctx_cancel_send, ctx)) != 0) {
-		nni_mtx_unlock(&s->lk);
-		nni_aio_finish_error(aio, rv);
-		return;
-	}
 	ctx->saio  = aio;
 	ctx->spipe = p;
 	nni_list_append(&p->sendq, ctx);
diff --git a/src/protocol/reqrep0/rep_test.c b/src/protocol/reqrep0/rep_test.c
index 7c6d6ba0..0eb8b985 100644
--- a/src/protocol/reqrep0/rep_test.c
+++ b/src/protocol/reqrep0/rep_test.c
@@ -410,6 +410,63 @@ test_rep_close_context_send(void)
 	TEST_NNG_PASS(nng_close(rep));
 }
 
+static void
+test_rep_ctx_recv_nonblock(void)
+{
+	nng_socket rep;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, rep));
+	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_ctx_recv(ctx, aio);
+
+	nng_aio_wait(aio);
+	TEST_NNG_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_close(rep));
+	nng_aio_free(aio);
+}
+
+static void
+test_rep_ctx_send_nonblock(void)
+{
+	nng_socket rep;
+	nng_socket req;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *msg;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, rep));
+	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	TEST_NNG_SEND_STR(req, "SEND");
+	nng_ctx_recv(ctx, aio);
+	nng_aio_wait(aio);
+	TEST_NNG_PASS(nng_aio_result(aio));
+	// message carries over
+	msg = nng_aio_get_msg(aio);
+	nng_aio_set_msg(aio, msg);
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_ctx_send(ctx, aio);
+
+	nng_aio_wait(aio);
+	TEST_NNG_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_close(rep));
+	TEST_NNG_PASS(nng_close(req));
+	nng_aio_free(aio);
+	nng_msg_free(msg);
+
+}
+
 void
 test_rep_recv_garbage(void)
 {
@@ -447,6 +504,8 @@ TEST_LIST = {
 	{ "rep recv aio ctx stopped", test_rep_ctx_recv_aio_stopped },
 	{ "rep close pipe context send", test_rep_close_pipe_context_send },
 	{ "rep close context send", test_rep_close_context_send },
+	{ "rep context send nonblock", test_rep_ctx_send_nonblock },
+	{ "rep context recv nonblock", test_rep_ctx_recv_nonblock },
 	{ "rep recv garbage", test_rep_recv_garbage },
 	{ NULL, NULL },
 };
diff --git a/src/protocol/reqrep0/req.c b/src/protocol/reqrep0/req.c
index 796bd71e..b5681688 100644
--- a/src/protocol/reqrep0/req.c
+++ b/src/protocol/reqrep0/req.c
@@ -134,17 +134,9 @@ static void
 req0_sock_close(void *arg)
 {
 	req0_sock *s = arg;
-	req0_ctx * ctx;
 
 	nni_mtx_lock(&s->mtx);
 	s->closed = true;
-	NNI_LIST_FOREACH (&s->contexts, ctx) {
-		if (ctx->recv_aio != NULL) {
-			nni_aio_finish_error(ctx->recv_aio, NNG_ECLOSED);
-			ctx->recv_aio = NULL;
-			req0_ctx_reset(ctx);
-		}
-	}
 	nni_mtx_unlock(&s->mtx);
 }
 
@@ -213,10 +205,6 @@ req0_pipe_start(void *arg)
 	}
 
 	nni_mtx_lock(&s->mtx);
-	if (s->closed || p->closed) {
-		nni_mtx_unlock(&s->mtx);
-		return (NNG_ECLOSED);
-	}
 	nni_list_append(&s->ready_pipes, p);
 	nni_pollable_raise(&s->writable);
 	req0_run_send_queue(s, NULL);
@@ -269,9 +257,9 @@ req0_send_cb(void *arg)
 	req0_pipe *p = arg;
 	req0_sock *s = p->req;
 	nni_aio *  aio;
-	nni_list   send_list;
+	nni_list   sent_list;
 
-	nni_aio_list_init(&send_list);
+	nni_aio_list_init(&sent_list);
 	if (nni_aio_result(&p->aio_send) != 0) {
 		// We failed to send... clean up and deal with it.
 		nni_msg_free(nni_aio_get_msg(&p->aio_send));
@@ -295,11 +283,11 @@ req0_send_cb(void *arg)
 	if (nni_list_empty(&s->send_queue)) {
 		nni_pollable_raise(&s->writable);
 	}
-	req0_run_send_queue(s, &send_list);
+	req0_run_send_queue(s, &sent_list);
 	nni_mtx_unlock(&s->mtx);
 
-	while ((aio = nni_list_first(&send_list)) != NULL) {
-		nni_list_remove(&send_list, aio);
+	while ((aio = nni_list_first(&sent_list)) != NULL) {
+		nni_list_remove(&sent_list, aio);
 		nni_aio_finish_synch(aio, 0, 0);
 	}
 }
@@ -451,7 +439,7 @@ req0_ctx_get_resend_time(void *arg, void *buf, size_t *szp, nni_opt_type t)
 }
 
 static void
-req0_run_send_queue(req0_sock *s, nni_list *send_list)
+req0_run_send_queue(req0_sock *s, nni_list *sent_list)
 {
 	req0_ctx *ctx;
 	nni_aio * aio;
@@ -495,24 +483,20 @@ req0_run_send_queue(req0_sock *s, nni_list *send_list)
 
 		nni_list_remove(&s->ready_pipes, p);
 		nni_list_append(&s->busy_pipes, p);
+		if (nni_list_empty(&s->ready_pipes)) {
+			nni_pollable_clear(&s->writable);
+		}
 
 		if ((aio = ctx->send_aio) != NULL) {
 			ctx->send_aio = NULL;
 			nni_aio_bump_count(aio, ctx->req_len);
 			// If the list was passed in, we want to do a
 			// synchronous completion later.
-			if (send_list != NULL) {
-				nni_list_append(send_list, aio);
+			if (sent_list != NULL) {
+				nni_list_append(sent_list, aio);
 			} else {
 				nni_aio_finish(aio, 0, 0);
 			}
-			if (ctx == &s->master) {
-				if (nni_list_empty(&s->ready_pipes)) {
-					nni_pollable_clear(&s->writable);
-				} else {
-					nni_pollable_raise(&s->writable);
-				}
-			}
 		}
 
 		nni_aio_set_msg(&p->aio_send, msg);
@@ -559,22 +543,20 @@ req0_ctx_cancel_recv(nni_aio *aio, void *arg, int rv)
 	req0_sock *s   = ctx->sock;
 
 	nni_mtx_lock(&s->mtx);
-	if (ctx->recv_aio != aio) {
-		// already completed, ignore this.
-		nni_mtx_unlock(&s->mtx);
-		return;
-	}
-	ctx->recv_aio = NULL;
+	if (ctx->recv_aio == aio) {
+		ctx->recv_aio = NULL;
 
-	// Cancellation of a pending receive is treated as aborting the
-	// entire state machine.  This allows us to preserve the semantic of
-	// exactly one receive operation per send operation, and should
-	// be the least surprising for users.  The main consequence is that
-	// if a receive operation is completed (in error or otherwise), the
-	// user must submit a new send operation to restart the state machine.
-	req0_ctx_reset(ctx);
+		// Cancellation of a pending receive is treated as aborting the
+		// entire state machine.  This allows us to preserve the
+		// semantic of exactly one receive operation per send
+		// operation, and should be the least surprising for users. The
+		// main consequence is that if a receive operation is completed
+		// (in error or otherwise), the user must submit a new send
+		// operation to restart the state machine.
+		req0_ctx_reset(ctx);
 
-	nni_aio_finish_error(aio, rv);
+		nni_aio_finish_error(aio, rv);
+	}
 	nni_mtx_unlock(&s->mtx);
 }
 
@@ -589,11 +571,6 @@ req0_ctx_recv(void *arg, nni_aio *aio)
 		return;
 	}
 	nni_mtx_lock(&s->mtx);
-	if (s->closed) {
-		nni_mtx_unlock(&s->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((ctx->recv_aio != NULL) ||
 	    ((ctx->req_msg == NULL) && (ctx->rep_msg == NULL))) {
 		// We have already got a pending receive or have not
@@ -635,30 +612,27 @@ req0_ctx_cancel_send(nni_aio *aio, void *arg, int rv)
 	req0_sock *s   = ctx->sock;
 
 	nni_mtx_lock(&s->mtx);
-	if (ctx->send_aio != aio) {
-		// already completed, ignore this.
-		nni_mtx_unlock(&s->mtx);
-		return;
-	}
+	if (ctx->send_aio == aio) {
+		// There should not be a pending reply, because we canceled
+		// it while we were waiting.
+		NNI_ASSERT(ctx->recv_aio == NULL);
+		ctx->send_aio = NULL;
+		// Restore the message back to the aio.
+		nni_aio_set_msg(aio, ctx->req_msg);
+		nni_msg_header_clear(ctx->req_msg);
+		ctx->req_msg = NULL;
 
-	// There should not be a pending reply, because we canceled
-	// it while we were waiting.
-	NNI_ASSERT(ctx->recv_aio == NULL);
-	ctx->send_aio = NULL;
-	// Restore the message back to the aio.
-	nni_aio_set_msg(aio, ctx->req_msg);
-	nni_msg_header_clear(ctx->req_msg);
-	ctx->req_msg = NULL;
-
-	// Cancellation of a pending receive is treated as aborting the
-	// entire state machine.  This allows us to preserve the semantic of
-	// exactly one receive operation per send operation, and should
-	// be the least surprising for users.  The main consequence is that
-	// if a receive operation is completed (in error or otherwise), the
-	// user must submit a new send operation to restart the state machine.
-	req0_ctx_reset(ctx);
+		// Cancellation of a pending receive is treated as aborting the
+		// entire state machine.  This allows us to preserve the
+		// semantic of exactly one receive operation per send
+		// operation, and should be the least surprising for users. The
+		// main consequence is that if a receive operation is completed
+		// (in error or otherwise), the user must submit a new send
+		// operation to restart the state machine.
+		req0_ctx_reset(ctx);
 
-	nni_aio_finish_error(aio, rv);
+		nni_aio_finish_error(aio, rv);
+	}
 	nni_mtx_unlock(&s->mtx);
 }
 
@@ -728,8 +702,6 @@ req0_ctx_send(void *arg, nni_aio *aio)
 	// Stick us on the send_queue list.
 	nni_list_append(&s->send_queue, ctx);
 
-	// Note that this will be synchronous if the ready_pipes list was
-	// not empty.
 	req0_run_send_queue(s, NULL);
 	nni_mtx_unlock(&s->mtx);
 }
diff --git a/src/protocol/reqrep0/req_test.c b/src/protocol/reqrep0/req_test.c
new file mode 100644
index 00000000..523af1d8
--- /dev/null
+++ b/src/protocol/reqrep0/req_test.c
@@ -0,0 +1,845 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2018 Capitar IT Group BV <info@capitar.com>
+//
+// 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 <string.h>
+
+#include <nng/nng.h>
+#include <nng/protocol/reqrep0/rep.h>
+#include <nng/protocol/reqrep0/req.h>
+
+#include <acutest.h>
+#include <testutil.h>
+
+#ifndef NNI_PROTO
+#define NNI_PROTO(x, y) (((x) << 4u) | (y))
+#endif
+
+static void
+test_req_identity(void)
+{
+	nng_socket s;
+	int        p;
+	char *     n;
+
+	TEST_NNG_PASS(nng_req0_open(&s));
+	TEST_NNG_PASS(nng_getopt_int(s, NNG_OPT_PROTO, &p));
+	TEST_CHECK(p == NNI_PROTO(3u, 0u)); // 48
+	TEST_NNG_PASS(nng_getopt_int(s, NNG_OPT_PEER, &p));
+	TEST_CHECK(p == NNI_PROTO(3u, 1u)); // 49
+	TEST_NNG_PASS(nng_getopt_string(s, NNG_OPT_PROTONAME, &n));
+	TEST_CHECK(strcmp(n, "req") == 0);
+	nng_strfree(n);
+	TEST_NNG_PASS(nng_getopt_string(s, NNG_OPT_PEERNAME, &n));
+	TEST_CHECK(strcmp(n, "rep") == 0);
+	nng_strfree(n);
+	TEST_NNG_PASS(nng_close(s));
+}
+
+static void
+test_req_ttl_option(void)
+{
+	nng_socket  req;
+	int         v;
+	bool        b;
+	size_t      sz;
+	const char *opt = NNG_OPT_MAXTTL;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+
+	TEST_NNG_PASS(nng_setopt_int(req, opt, 1));
+	TEST_NNG_FAIL(nng_setopt_int(req, opt, 0), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_int(req, opt, -1), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_int(req, opt, 256), NNG_EINVAL);
+	TEST_NNG_PASS(nng_setopt_int(req, opt, 3));
+	TEST_NNG_PASS(nng_getopt_int(req, opt, &v));
+	TEST_CHECK(v == 3);
+	v  = 0;
+	sz = sizeof(v);
+	TEST_NNG_PASS(nng_getopt(req, opt, &v, &sz));
+	TEST_CHECK(v == 3);
+	TEST_CHECK(sz == sizeof(v));
+
+	TEST_NNG_FAIL(nng_setopt(req, opt, "", 1), NNG_EINVAL);
+	sz = 1;
+	TEST_NNG_FAIL(nng_getopt(req, opt, &v, &sz), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_bool(req, opt, true), NNG_EBADTYPE);
+	TEST_NNG_FAIL(nng_getopt_bool(req, opt, &b), NNG_EBADTYPE);
+
+	TEST_NNG_PASS(nng_close(req));
+}
+
+static void
+test_req_resend_option(void)
+{
+	nng_socket   req;
+	nng_duration d;
+	bool         b;
+	size_t       sz  = sizeof(b);
+	const char * opt = NNG_OPT_REQ_RESENDTIME;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+
+	TEST_CHECK(nng_setopt_ms(req, opt, 10) == 0);
+	TEST_NNG_FAIL(nng_setopt(req, opt, "", 1), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_getopt(req, opt, &b, &sz), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_bool(req, opt, true), NNG_EBADTYPE);
+	TEST_NNG_FAIL(nng_getopt_bool(req, opt, &b), NNG_EBADTYPE);
+
+	TEST_NNG_PASS(nng_getopt_ms(req, opt, &d));
+	TEST_CHECK(d == 10);
+	TEST_NNG_PASS(nng_close(req));
+}
+
+void
+test_req_recv_bad_state(void)
+{
+	nng_socket req;
+	nng_msg *  msg = NULL;
+
+	TEST_CHECK(nng_req0_open(&req) == 0);
+	TEST_CHECK(nng_recvmsg(req, &msg, 0) == NNG_ESTATE);
+	TEST_CHECK(msg == NULL);
+	TEST_CHECK(nng_close(req) == 0);
+}
+
+static void
+test_req_recv_garbage(void)
+{
+	nng_socket rep;
+	nng_socket req;
+	nng_msg *  m;
+	uint32_t   req_id;
+
+	TEST_NNG_PASS(nng_rep0_open_raw(&rep));
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, 100));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	TEST_NNG_PASS(nng_msg_alloc(&m, 0));
+	TEST_NNG_PASS(nng_sendmsg(req, m, 0));
+
+	TEST_NNG_PASS(nng_recvmsg(rep, &m, 0));
+
+	// The message will have a header that contains the 32-bit pipe ID,
+	// followed by the 32-bit request ID.  We will discard the request
+	// ID before sending it out.
+	TEST_CHECK(nng_msg_header_len(m) == 8);
+	TEST_NNG_PASS(nng_msg_header_chop_u32(m, &req_id));
+
+	TEST_NNG_PASS(nng_sendmsg(rep, m, 0));
+	TEST_NNG_FAIL(nng_recvmsg(req, &m, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+}
+
+#define SECOND 1000
+
+void
+test_req_rep_exchange(void)
+{
+	nng_socket req;
+	nng_socket rep;
+	nng_msg *  msg = NULL;
+
+	TEST_CHECK(nng_req0_open(&req) == 0);
+	TEST_CHECK(nng_rep0_open(&rep) == 0);
+
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, SECOND) == 0);
+
+	TEST_CHECK(testutil_marry(rep, req) == 0);
+
+	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
+	TEST_CHECK(nng_msg_append(msg, "ping", 5) == 0);
+	TEST_CHECK(nng_msg_len(msg) == 5);
+	TEST_CHECK(strcmp(nng_msg_body(msg), "ping") == 0);
+	TEST_CHECK(nng_sendmsg(req, msg, 0) == 0);
+	msg = NULL;
+	TEST_CHECK(nng_recvmsg(rep, &msg, 0) == 0);
+	TEST_CHECK(msg != NULL);
+	TEST_CHECK(nng_msg_len(msg) == 5);
+	TEST_CHECK(strcmp(nng_msg_body(msg), "ping") == 0);
+	nng_msg_trim(msg, 5);
+	TEST_CHECK(nng_msg_append(msg, "pong", 5) == 0);
+	TEST_CHECK(nng_sendmsg(rep, msg, 0) == 0);
+	msg = NULL;
+	TEST_CHECK(nng_recvmsg(req, &msg, 0) == 0);
+	TEST_CHECK(msg != NULL);
+	TEST_CHECK(nng_msg_len(msg) == 5);
+	TEST_CHECK(strcmp(nng_msg_body(msg), "pong") == 0);
+	nng_msg_free(msg);
+
+	TEST_CHECK(nng_close(req) == 0);
+	TEST_CHECK(nng_close(rep) == 0);
+}
+
+void
+test_req_cancel(void)
+{
+	nng_msg *    abc;
+	nng_msg *    def;
+	nng_msg *    cmd;
+	nng_duration retry = SECOND;
+	nng_socket   req;
+	nng_socket   rep;
+
+	TEST_CHECK(nng_rep_open(&rep) == 0);
+	TEST_CHECK(nng_req_open(&req) == 0);
+
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_REQ_RESENDTIME, retry) == 0);
+	TEST_CHECK(nng_setopt_int(req, NNG_OPT_SENDBUF, 16) == 0);
+
+	TEST_CHECK(nng_msg_alloc(&abc, 0) == 0);
+	TEST_CHECK(nng_msg_append(abc, "abc", 4) == 0);
+	TEST_CHECK(nng_msg_alloc(&def, 0) == 0);
+	TEST_CHECK(nng_msg_append(def, "def", 4) == 0);
+
+	TEST_CHECK(testutil_marry(rep, req) == 0);
+
+	// Send req #1 (abc).
+	TEST_CHECK(nng_sendmsg(req, abc, 0) == 0);
+
+	// Sleep a bit.  This is so that we ensure that our request gets
+	// to the far side.  (If we cancel too fast, then our outgoing send
+	// will be canceled before it gets to the peer.)
+	testutil_sleep(100);
+
+	// Send the next next request ("def").  Note that
+	// the REP side server will have already buffered the receive
+	// request, and should simply be waiting for us to reply to abc.
+	TEST_CHECK(nng_sendmsg(req, def, 0) == 0);
+
+	// Receive the first request (should be abc) on the REP server.
+	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
+	TEST_ASSERT(cmd != NULL);
+	TEST_CHECK(nng_msg_len(cmd) == 4);
+	TEST_CHECK(strcmp(nng_msg_body(cmd), "abc") == 0);
+
+	// REP sends the reply to first command.  This will be discarded
+	// by the REQ socket.
+	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
+
+	// Now get the next command from the REP; should be "def".
+	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
+	TEST_ASSERT(cmd != NULL);
+	TEST_CHECK(nng_msg_len(cmd) == 4);
+	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
+	TEST_MSG("Received body was %s", nng_msg_body(cmd));
+
+	// And send it back to REQ.
+	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
+
+	// Try a req command.  This should give back "def"
+	TEST_CHECK(nng_recvmsg(req, &cmd, 0) == 0);
+	TEST_CHECK(nng_msg_len(cmd) == 4);
+	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
+	nng_msg_free(cmd);
+
+	TEST_CHECK(nng_close(req) == 0);
+	TEST_CHECK(nng_close(rep) == 0);
+}
+
+void
+test_req_cancel_abort_recv(void)
+{
+
+	nng_msg *    abc;
+	nng_msg *    def;
+	nng_msg *    cmd;
+	nng_aio *    aio;
+	nng_duration retry = SECOND * 10; // 10s (kind of never)
+	nng_socket   req;
+	nng_socket   rep;
+
+	TEST_CHECK(nng_rep_open(&rep) == 0);
+	TEST_CHECK(nng_req_open(&req) == 0);
+	TEST_CHECK(nng_aio_alloc(&aio, NULL, NULL) == 0);
+
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_REQ_RESENDTIME, retry) == 0);
+	TEST_CHECK(nng_setopt_int(req, NNG_OPT_SENDBUF, 16) == 0);
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, 5 * SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 5 * SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 5 * SECOND) == 0);
+	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 5 * SECOND) == 0);
+
+	TEST_CHECK(nng_msg_alloc(&abc, 0) == 0);
+	TEST_CHECK(nng_msg_append(abc, "abc", 4) == 0);
+	TEST_CHECK(nng_msg_alloc(&def, 0) == 0);
+	TEST_CHECK(nng_msg_append(def, "def", 4) == 0);
+
+	TEST_CHECK(testutil_marry(rep, req) == 0);
+
+	// Send req #1 (abc).
+	TEST_CHECK(nng_sendmsg(req, abc, 0) == 0);
+
+	// Wait for it to get ot the other side.
+	testutil_sleep(100);
+
+	nng_aio_set_timeout(aio, 5 * SECOND);
+	nng_recv_aio(req, aio);
+
+	// Give time for this recv to post properly.
+	testutil_sleep(100);
+
+	// Send the next next request ("def").  Note that
+	// the REP side server will have already buffered the receive
+	// request, and should simply be waiting for us to reply to
+	// abc.
+	TEST_CHECK(nng_sendmsg(req, def, 0) == 0);
+
+	// Our pending I/O should have been canceled.
+	nng_aio_wait(aio);
+	TEST_CHECK(nng_aio_result(aio) == NNG_ECANCELED);
+
+	// Receive the first request (should be abc) on the REP server.
+	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
+	TEST_CHECK(nng_msg_len(cmd) == 4);
+	TEST_CHECK(strcmp(nng_msg_body(cmd), "abc") == 0);
+
+	// REP sends the reply to first command.  This will be
+	// discarded by the REQ socket.
+	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
+
+	// Now get the next command from the REP; should be "def".
+	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
+	TEST_CHECK(nng_msg_len(cmd) == 4);
+	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
+
+	// And send it back to REQ.
+	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
+
+	// Try a req command.  This should give back "def"
+	TEST_CHECK(nng_recvmsg(req, &cmd, 0) == 0);
+	TEST_CHECK(nng_msg_len(cmd) == 4);
+	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
+	nng_msg_free(cmd);
+
+	nng_aio_free(aio);
+	TEST_CHECK(nng_close(req) == 0);
+	TEST_CHECK(nng_close(rep) == 0);
+}
+
+static void
+test_req_cancel_post_recv(void)
+{
+	nng_socket req;
+	nng_socket rep;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000));
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	TEST_NNG_SEND_STR(req, "ONE");
+	TEST_NNG_RECV_STR(rep, "ONE");
+	TEST_NNG_SEND_STR(rep, "one");
+	testutil_sleep(100); // Make sure reply arrives!
+	TEST_NNG_SEND_STR(req, "TWO");
+	TEST_NNG_RECV_STR(rep, "TWO");
+	TEST_NNG_SEND_STR(rep, "two");
+	TEST_NNG_RECV_STR(req, "two");
+
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+}
+
+void
+test_req_poll_writeable(void)
+{
+	int        fd;
+	nng_socket req;
+	nng_socket rep;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_SENDFD, &fd));
+	TEST_CHECK(fd >= 0);
+
+	// Not writable before connect.
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	// It should be writable now.
+	TEST_CHECK(testutil_pollfd(fd) == true);
+
+	// Submit a bunch of jobs.  Note that we have to stall a bit
+	// between each message to let it queue up.
+	for (int i = 0; i < 10; i++) {
+		int rv = nng_send(req, "", 0, NNG_FLAG_NONBLOCK);
+		if (rv == NNG_EAGAIN) {
+			break;
+		}
+		TEST_NNG_PASS(rv);
+		testutil_sleep(50);
+	}
+	TEST_CHECK(testutil_pollfd(fd) == 0);
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+}
+
+void
+test_req_poll_contention(void)
+{
+	int        fd;
+	nng_socket req;
+	nng_socket rep;
+	nng_aio *  aio;
+	nng_ctx    ctx[5];
+	nng_aio *  ctx_aio[5];
+	nng_msg *  ctx_msg[5];
+	nng_msg *  msg;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_setopt_int(req, NNG_OPT_SENDBUF, 1));
+	TEST_NNG_PASS(nng_setopt_int(rep, NNG_OPT_RECVBUF, 1));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000));
+
+	for (int i = 0; i < 5; i++) {
+		TEST_NNG_PASS(nng_ctx_open(&ctx[i], req));
+		TEST_NNG_PASS(nng_aio_alloc(&ctx_aio[i], NULL, NULL));
+		TEST_NNG_PASS(nng_msg_alloc(&ctx_msg[i], 0));
+	}
+	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+
+	TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_SENDFD, &fd));
+	TEST_CHECK(fd >= 0);
+
+	// Not writable before connect.
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	nng_aio_set_msg(aio, msg);
+	nng_send_aio(req, aio);
+	for (int i = 0; i < 5; i++) {
+		nng_aio_set_msg(ctx_aio[i], ctx_msg[i]);
+		nng_ctx_send(ctx[i], ctx_aio[i]);
+	}
+	testutil_sleep(50); // so everything is queued steady state
+
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	// It should not be writable now.
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0));
+	nng_msg_free(msg);
+
+	// Still not writeable...
+	TEST_CHECK(testutil_pollfd(fd) == false);
+	for (int i = 0; i < 5; i++) {
+		TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0));
+		nng_msg_free(msg);
+	}
+	// Should be come writeable now...
+	TEST_CHECK(testutil_pollfd(fd) == true);
+
+	for (int i = 0; i < 5; i++) {
+		nng_aio_free(ctx_aio[i]);
+	}
+	nng_aio_free(aio);
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+}
+
+void
+test_req_poll_multi_pipe(void)
+{
+	int        fd;
+	nng_socket req;
+	nng_socket rep1;
+	nng_socket rep2;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep1));
+	TEST_NNG_PASS(nng_rep0_open(&rep2));
+	TEST_NNG_PASS(nng_setopt_int(req, NNG_OPT_SENDBUF, 1));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+
+	TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_SENDFD, &fd));
+	TEST_CHECK(fd >= 0);
+
+	// Not writable before connect.
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	TEST_NNG_PASS(testutil_marry(req, rep1));
+	TEST_NNG_PASS(testutil_marry(req, rep2));
+
+	TEST_CHECK(testutil_pollfd(fd) == true);
+	TEST_NNG_SEND_STR(req, "ONE");
+	TEST_CHECK(testutil_pollfd(fd) == true);
+
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep1));
+	TEST_NNG_PASS(nng_close(rep2));
+}
+
+void
+test_req_poll_readable(void)
+{
+	int        fd;
+	nng_socket req;
+	nng_socket rep;
+	nng_msg *  msg;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_RECVFD, &fd));
+	TEST_CHECK(fd >= 0);
+
+	// Not readable if not connected!
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	// Even after connect (no message yet)
+	TEST_NNG_PASS(testutil_marry(req, rep));
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	// But once we send messages, it is.
+	// We have to send a request, in order to send a reply.
+
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append(msg, "xyz", 3));
+	TEST_NNG_PASS(nng_sendmsg(req, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0)); // recv on rep
+	TEST_NNG_PASS(nng_sendmsg(rep, msg, 0));  // echo it back
+	testutil_sleep(200); // give time for message to arrive
+
+	TEST_CHECK(testutil_pollfd(fd) == true);
+
+	// and receiving makes it no longer ready
+	TEST_NNG_PASS(nng_recvmsg(req, &msg, 0));
+	nng_msg_free(msg);
+	TEST_CHECK(testutil_pollfd(fd) == false);
+
+	// TODO verify unsolicited response
+
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+}
+
+static void
+test_req_ctx_no_poll(void)
+{
+	int        fd;
+	nng_socket req;
+	nng_ctx    ctx;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, req));
+	TEST_NNG_FAIL(
+	    nng_ctx_getopt_int(ctx, NNG_OPT_SENDFD, &fd), NNG_ENOTSUP);
+	TEST_NNG_FAIL(
+	    nng_ctx_getopt_int(ctx, NNG_OPT_RECVFD, &fd), NNG_ENOTSUP);
+	TEST_NNG_PASS(nng_ctx_close(ctx));
+	TEST_NNG_PASS(nng_close(req));
+}
+
+static void
+test_req_ctx_send_queued(void)
+{
+	nng_socket req;
+	nng_socket rep;
+	nng_ctx    ctx[3];
+	nng_aio *  aio[3];
+	nng_msg *  msg[3];
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 100));
+
+	for (int i = 0; i < 3; i++) {
+		TEST_NNG_PASS(nng_ctx_open(&ctx[i], req));
+		TEST_NNG_PASS(nng_aio_alloc(&aio[i], NULL, NULL));
+		TEST_NNG_PASS(nng_msg_alloc(&msg[i], 0));
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_aio_set_msg(aio[i], msg[i]);
+		nng_ctx_send(ctx[i], aio[i]);
+	}
+
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	testutil_sleep(50); // Only to ensure stuff queues up
+	for (int i = 0; i < 3; i++) {
+		nng_msg *m;
+		TEST_NNG_PASS(nng_recvmsg(rep, &m, 0));
+		nng_msg_free(m);
+	}
+
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+	for (int i = 0; i < 3; i++) {
+		nng_aio_wait(aio[i]);
+		TEST_NNG_PASS(nng_aio_result(aio[i]));
+		nng_aio_free(aio[i]);
+	}
+}
+
+static void
+test_req_ctx_send_close(void)
+{
+	nng_socket req;
+	nng_ctx    ctx[3];
+	nng_aio *  aio[3];
+	nng_msg *  msg[3];
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+
+	for (int i = 0; i < 3; i++) {
+		TEST_NNG_PASS(nng_ctx_open(&ctx[i], req));
+		TEST_NNG_PASS(nng_aio_alloc(&aio[i], NULL, NULL));
+		TEST_NNG_PASS(nng_msg_alloc(&msg[i], 0));
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_aio_set_msg(aio[i], msg[i]);
+		nng_ctx_send(ctx[i], aio[i]);
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_ctx_close(ctx[i]);
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_aio_wait(aio[i]);
+		TEST_NNG_FAIL(nng_aio_result(aio[i]), NNG_ECLOSED);
+		nng_aio_free(aio[i]);
+		nng_msg_free(msg[i]);
+	}
+	TEST_NNG_PASS(nng_close(req));
+}
+
+static void
+test_req_ctx_send_abort(void)
+{
+	nng_socket req;
+	nng_ctx    ctx[3];
+	nng_aio *  aio[3];
+	nng_msg *  msg[3];
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+
+	for (int i = 0; i < 3; i++) {
+		TEST_NNG_PASS(nng_ctx_open(&ctx[i], req));
+		TEST_NNG_PASS(nng_aio_alloc(&aio[i], NULL, NULL));
+		TEST_NNG_PASS(nng_msg_alloc(&msg[i], 0));
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_aio_set_msg(aio[i], msg[i]);
+		nng_ctx_send(ctx[i], aio[i]);
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_aio_abort(aio[i], NNG_ECANCELED);
+	}
+
+	for (int i = 0; i < 3; i++) {
+		nng_aio_wait(aio[i]);
+		TEST_NNG_FAIL(nng_aio_result(aio[i]), NNG_ECANCELED);
+		nng_aio_free(aio[i]);
+		nng_msg_free(msg[i]);
+	}
+	TEST_NNG_PASS(nng_close(req));
+}
+
+static void
+test_req_ctx_send_twice(void)
+{
+	nng_socket req;
+	nng_ctx    ctx;
+	nng_aio *  aio[2];
+	nng_msg *  msg[2];
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, req));
+
+	for (int i = 0; i < 2; i++) {
+		TEST_NNG_PASS(nng_aio_alloc(&aio[i], NULL, NULL));
+		TEST_NNG_PASS(nng_msg_alloc(&msg[i], 0));
+	}
+
+	for (int i = 0; i < 2; i++) {
+		nng_aio_set_msg(aio[i], msg[i]);
+		nng_ctx_send(ctx, aio[i]);
+		testutil_sleep(50);
+	}
+
+	TEST_NNG_PASS(nng_close(req));
+	nng_aio_wait(aio[0]);
+	nng_aio_wait(aio[1]);
+	TEST_NNG_FAIL(nng_aio_result(aio[0]), NNG_ECANCELED);
+	TEST_NNG_FAIL(nng_aio_result(aio[1]), NNG_ECLOSED);
+
+	for (int i = 0; i < 2; i++) {
+		nng_aio_free(aio[i]);
+		nng_msg_free(msg[i]);
+	}
+}
+
+static void
+test_req_ctx_recv_nonblock(void)
+{
+	nng_socket req;
+	nng_socket rep;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  msg;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, req));
+	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+
+	TEST_NNG_PASS(testutil_marry(req, rep));
+
+	nng_aio_set_msg(aio, msg);
+	nng_ctx_send(ctx, aio);
+	nng_aio_wait(aio);
+	TEST_NNG_PASS(nng_aio_result(aio));
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_ctx_recv(ctx, aio);
+
+	nng_aio_wait(aio);
+	TEST_NNG_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_close(req));
+	TEST_NNG_PASS(nng_close(rep));
+	nng_aio_free(aio);
+}
+
+static void
+test_req_ctx_send_nonblock(void)
+{
+	nng_socket req;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  msg;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, req));
+	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+
+	nng_aio_set_msg(aio, msg);
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_ctx_send(ctx, aio);
+	nng_aio_wait(aio);
+	TEST_NNG_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_close(req));
+	nng_aio_free(aio);
+	nng_msg_free(msg);
+}
+
+static void
+test_req_ctx_recv_close_socket(void)
+{
+	nng_socket req;
+	nng_socket rep;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	TEST_NNG_PASS(nng_req0_open(&req));
+	TEST_NNG_PASS(nng_rep0_open(&rep));
+	TEST_NNG_PASS(nng_ctx_open(&ctx, req));
+	TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	TEST_NNG_PASS(testutil_marry(req, rep));
+	TEST_NNG_PASS(nng_msg_alloc(&m, 0));
+	nng_aio_set_msg(aio, m);
+	nng_ctx_send(ctx, aio);
+	nng_aio_wait(aio);
+	TEST_NNG_PASS(nng_aio_result(aio));
+
+	nng_ctx_recv(ctx, aio);
+	nng_close(req);
+
+	TEST_NNG_FAIL(nng_aio_result(aio), NNG_ECLOSED);
+	nng_aio_free(aio);
+	TEST_NNG_PASS(nng_close(rep));
+}
+
+static void
+test_req_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char       addr[64];
+
+	testutil_scratch_addr("inproc", sizeof(addr), addr);
+
+	TEST_NNG_PASS(nng_req0_open(&s1));
+	TEST_NNG_PASS(nng_req0_open(&s2));
+
+	TEST_NNG_PASS(nng_listen(s1, addr, NULL, 0));
+	TEST_NNG_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	testutil_sleep(100);
+	TEST_NNG_PASS(nng_stats_get(&stats));
+
+	TEST_CHECK(stats != NULL);
+	TEST_CHECK((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	TEST_CHECK((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	TEST_CHECK(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	TEST_CHECK(nng_stat_value(reject) > 0);
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(s2));
+	nng_stats_free(stats);
+}
+
+TEST_LIST = {
+	{ "req rep identity", test_req_identity },
+	{ "req resend option", test_req_ttl_option },
+	{ "req resend option", test_req_resend_option },
+	{ "req recv bad state", test_req_recv_bad_state },
+	{ "req recv garbage", test_req_recv_garbage },
+	{ "req rep exchange", test_req_rep_exchange },
+	{ "req cancel", test_req_cancel },
+	{ "req cancel abort recv", test_req_cancel_abort_recv },
+	{ "req cancel post recv", test_req_cancel_post_recv },
+	{ "req poll writable", test_req_poll_writeable },
+	{ "req poll contention", test_req_poll_contention },
+	{ "req poll multi pipe", test_req_poll_multi_pipe },
+	{ "req poll readable", test_req_poll_readable },
+	{ "req context send queued", test_req_ctx_send_queued },
+	{ "req context send close", test_req_ctx_send_close },
+	{ "req context send abort", test_req_ctx_send_abort },
+	{ "req context send twice", test_req_ctx_send_twice },
+	{ "req context does not poll", test_req_ctx_no_poll },
+	{ "req context recv close socket", test_req_ctx_recv_close_socket },
+	{ "req context recv nonblock", test_req_ctx_recv_nonblock },
+	{ "req context send nonblock", test_req_ctx_send_nonblock },
+	{ "req validate peer", test_req_validate_peer },
+	{ NULL, NULL },
+};
diff --git a/src/protocol/reqrep0/reqrep_test.c b/src/protocol/reqrep0/reqrep_test.c
deleted file mode 100644
index f4617936..00000000
--- a/src/protocol/reqrep0/reqrep_test.c
+++ /dev/null
@@ -1,434 +0,0 @@
-//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
-// Copyright 2018 Capitar IT Group BV <info@capitar.com>
-//
-// 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 <string.h>
-
-#include <nng/nng.h>
-#include <nng/protocol/reqrep0/rep.h>
-#include <nng/protocol/reqrep0/req.h>
-
-#include <acutest.h>
-#include <testutil.h>
-
-#ifndef NNI_PROTO
-#define NNI_PROTO(x, y) (((x) << 4u) | (y))
-#endif
-
-void
-test_req_rep_identity(void)
-{
-	nng_socket s;
-	int        p;
-	char *     n;
-
-	TEST_CHECK(nng_req0_open(&s) == 0);
-	TEST_CHECK(nng_getopt_int(s, NNG_OPT_PROTO, &p) == 0);
-	TEST_CHECK(p == NNI_PROTO(3u, 0u)); // 48
-	TEST_CHECK(nng_getopt_int(s, NNG_OPT_PEER, &p) == 0);
-	TEST_CHECK(p == NNI_PROTO(3u, 1u)); // 49
-	TEST_CHECK(nng_getopt_string(s, NNG_OPT_PROTONAME, &n) == 0);
-	TEST_CHECK(strcmp(n, "req") == 0);
-	nng_strfree(n);
-	TEST_CHECK(nng_getopt_string(s, NNG_OPT_PEERNAME, &n) == 0);
-	TEST_CHECK(strcmp(n, "rep") == 0);
-	nng_strfree(n);
-	TEST_CHECK(nng_close(s) == 0);
-
-	TEST_CHECK(nng_rep0_open(&s) == 0);
-	TEST_CHECK(nng_getopt_int(s, NNG_OPT_PROTO, &p) == 0);
-	TEST_CHECK(p == NNI_PROTO(3u, 1u)); // 49
-	TEST_CHECK(nng_getopt_int(s, NNG_OPT_PEER, &p) == 0);
-	TEST_CHECK(p == NNI_PROTO(3u, 0u)); // 48
-	TEST_CHECK(nng_getopt_string(s, NNG_OPT_PROTONAME, &n) == 0);
-	TEST_CHECK(strcmp(n, "rep") == 0);
-	nng_strfree(n);
-	TEST_CHECK(nng_getopt_string(s, NNG_OPT_PEERNAME, &n) == 0);
-	TEST_CHECK(strcmp(n, "req") == 0);
-	nng_strfree(n);
-	TEST_CHECK(nng_close(s) == 0);
-}
-
-void
-test_resend_option(void)
-{
-	nng_socket  req;
-	bool        b;
-	size_t      sz  = sizeof(b);
-	const char *opt = NNG_OPT_REQ_RESENDTIME;
-
-	TEST_CHECK(nng_req0_open(&req) == 0);
-
-	TEST_CHECK(nng_setopt_ms(req, opt, 10) == 0);
-	TEST_CHECK(nng_setopt(req, opt, "", 1) == NNG_EINVAL);
-	TEST_CHECK(nng_getopt(req, opt, &b, &sz) == NNG_EINVAL);
-	TEST_CHECK(nng_setopt_bool(req, opt, true) == NNG_EBADTYPE);
-	TEST_CHECK(nng_getopt_bool(req, opt, &b) == NNG_EBADTYPE);
-
-	TEST_CHECK(nng_close(req) == 0);
-}
-
-void
-test_req_recv_bad_state(void)
-{
-	nng_socket req;
-	nng_msg *  msg = NULL;
-
-	TEST_CHECK(nng_req0_open(&req) == 0);
-	TEST_CHECK(nng_recvmsg(req, &msg, 0) == NNG_ESTATE);
-	TEST_CHECK(msg == NULL);
-	TEST_CHECK(nng_close(req) == 0);
-}
-
-void
-test_rep_send_bad_state(void)
-{
-	nng_socket rep;
-	nng_msg *  msg = NULL;
-
-	TEST_CHECK(nng_rep0_open(&rep) == 0);
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_sendmsg(rep, msg, 0) == NNG_ESTATE);
-	nng_msg_free(msg);
-	TEST_CHECK(nng_close(rep) == 0);
-}
-
-#define SECOND 1000
-
-void
-test_req_rep_exchange(void)
-{
-	nng_socket req;
-	nng_socket rep;
-	nng_msg *  msg = NULL;
-
-	TEST_CHECK(nng_req0_open(&req) == 0);
-	TEST_CHECK(nng_rep0_open(&rep) == 0);
-
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, SECOND) == 0);
-
-	TEST_CHECK(testutil_marry(rep, req) == 0);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_append(msg, "ping", 5) == 0);
-	TEST_CHECK(nng_msg_len(msg) == 5);
-	TEST_CHECK(strcmp(nng_msg_body(msg), "ping") == 0);
-	TEST_CHECK(nng_sendmsg(req, msg, 0) == 0);
-	msg = NULL;
-	TEST_CHECK(nng_recvmsg(rep, &msg, 0) == 0);
-	TEST_CHECK(msg != NULL);
-	TEST_CHECK(nng_msg_len(msg) == 5);
-	TEST_CHECK(strcmp(nng_msg_body(msg), "ping") == 0);
-	nng_msg_trim(msg, 5);
-	TEST_CHECK(nng_msg_append(msg, "pong", 5) == 0);
-	TEST_CHECK(nng_sendmsg(rep, msg, 0) == 0);
-	msg = NULL;
-	TEST_CHECK(nng_recvmsg(req, &msg, 0) == 0);
-	TEST_CHECK(msg != NULL);
-	TEST_CHECK(nng_msg_len(msg) == 5);
-	TEST_CHECK(strcmp(nng_msg_body(msg), "pong") == 0);
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_close(req) == 0);
-	TEST_CHECK(nng_close(rep) == 0);
-}
-
-void
-test_req_cancel(void)
-{
-	nng_msg *    abc;
-	nng_msg *    def;
-	nng_msg *    cmd;
-	nng_duration retry = SECOND;
-	nng_socket   req;
-	nng_socket   rep;
-
-	TEST_CHECK(nng_rep_open(&rep) == 0);
-	TEST_CHECK(nng_req_open(&req) == 0);
-
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_REQ_RESENDTIME, retry) == 0);
-	TEST_CHECK(nng_setopt_int(req, NNG_OPT_SENDBUF, 16) == 0);
-
-	TEST_CHECK(nng_msg_alloc(&abc, 0) == 0);
-	TEST_CHECK(nng_msg_append(abc, "abc", 4) == 0);
-	TEST_CHECK(nng_msg_alloc(&def, 0) == 0);
-	TEST_CHECK(nng_msg_append(def, "def", 4) == 0);
-
-	TEST_CHECK(testutil_marry(rep, req) == 0);
-
-	// Send req #1 (abc).
-	TEST_CHECK(nng_sendmsg(req, abc, 0) == 0);
-
-	// Sleep a bit.  This is so that we ensure that our request gets
-	// to the far side.  (If we cancel too fast, then our outgoing send
-	// will be canceled before it gets to the peer.)
-	testutil_sleep(100);
-
-	// Send the next next request ("def").  Note that
-	// the REP side server will have already buffered the receive
-	// request, and should simply be waiting for us to reply to abc.
-	TEST_CHECK(nng_sendmsg(req, def, 0) == 0);
-
-	// Receive the first request (should be abc) on the REP server.
-	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
-	TEST_ASSERT(cmd != NULL);
-	TEST_CHECK(nng_msg_len(cmd) == 4);
-	TEST_CHECK(strcmp(nng_msg_body(cmd), "abc") == 0);
-
-	// REP sends the reply to first command.  This will be discarded
-	// by the REQ socket.
-	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
-
-	// Now get the next command from the REP; should be "def".
-	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
-	TEST_ASSERT(cmd != NULL);
-	TEST_CHECK(nng_msg_len(cmd) == 4);
-	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
-	TEST_MSG("Received body was %s", nng_msg_body(cmd));
-
-	// And send it back to REQ.
-	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
-
-	// Try a req command.  This should give back "def"
-	TEST_CHECK(nng_recvmsg(req, &cmd, 0) == 0);
-	TEST_CHECK(nng_msg_len(cmd) == 4);
-	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
-	nng_msg_free(cmd);
-
-	TEST_CHECK(nng_close(req) == 0);
-	TEST_CHECK(nng_close(rep) == 0);
-}
-
-void
-test_req_cancel_abort_recv(void)
-{
-
-	nng_msg *    abc;
-	nng_msg *    def;
-	nng_msg *    cmd;
-	nng_aio *    aio;
-	nng_duration retry = SECOND * 10; // 10s (kind of never)
-	nng_socket   req;
-	nng_socket   rep;
-
-	TEST_CHECK(nng_rep_open(&rep) == 0);
-	TEST_CHECK(nng_req_open(&req) == 0);
-	TEST_CHECK(nng_aio_alloc(&aio, NULL, NULL) == 0);
-
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_REQ_RESENDTIME, retry) == 0);
-	TEST_CHECK(nng_setopt_int(req, NNG_OPT_SENDBUF, 16) == 0);
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, 5 * SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 5 * SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 5 * SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 5 * SECOND) == 0);
-
-	TEST_CHECK(nng_msg_alloc(&abc, 0) == 0);
-	TEST_CHECK(nng_msg_append(abc, "abc", 4) == 0);
-	TEST_CHECK(nng_msg_alloc(&def, 0) == 0);
-	TEST_CHECK(nng_msg_append(def, "def", 4) == 0);
-
-	TEST_CHECK(testutil_marry(rep, req) == 0);
-
-	// Send req #1 (abc).
-	TEST_CHECK(nng_sendmsg(req, abc, 0) == 0);
-
-	// Wait for it to get ot the other side.
-	testutil_sleep(100);
-
-	nng_aio_set_timeout(aio, 5 * SECOND);
-	nng_recv_aio(req, aio);
-
-	// Give time for this recv to post properly.
-	testutil_sleep(100);
-
-	// Send the next next request ("def").  Note that
-	// the REP side server will have already buffered the receive
-	// request, and should simply be waiting for us to reply to
-	// abc.
-	TEST_CHECK(nng_sendmsg(req, def, 0) == 0);
-
-	// Our pending I/O should have been canceled.
-	nng_aio_wait(aio);
-	TEST_CHECK(nng_aio_result(aio) == NNG_ECANCELED);
-
-	// Receive the first request (should be abc) on the REP server.
-	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
-	TEST_CHECK(nng_msg_len(cmd) == 4);
-	TEST_CHECK(strcmp(nng_msg_body(cmd), "abc") == 0);
-
-	// REP sends the reply to first command.  This will be
-	// discarded by the REQ socket.
-	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
-
-	// Now get the next command from the REP; should be "def".
-	TEST_CHECK(nng_recvmsg(rep, &cmd, 0) == 0);
-	TEST_CHECK(nng_msg_len(cmd) == 4);
-	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
-
-	// And send it back to REQ.
-	TEST_CHECK(nng_sendmsg(rep, cmd, 0) == 0);
-
-	// Try a req command.  This should give back "def"
-	TEST_CHECK(nng_recvmsg(req, &cmd, 0) == 0);
-	TEST_CHECK(nng_msg_len(cmd) == 4);
-	TEST_CHECK(strcmp(nng_msg_body(cmd), "def") == 0);
-	nng_msg_free(cmd);
-
-	nng_aio_free(aio);
-	TEST_CHECK(nng_close(req) == 0);
-	TEST_CHECK(nng_close(rep) == 0);
-}
-
-void
-test_req_poll_writeable(void)
-{
-	int        fd;
-	nng_socket req;
-	nng_socket rep;
-
-	TEST_NNG_PASS(nng_req0_open(&req));
-	TEST_NNG_PASS(nng_rep0_open(&rep));
-	TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_SENDFD, &fd));
-	TEST_CHECK(fd >= 0);
-
-	// Not writable before connect.
-	TEST_CHECK(testutil_pollfd(fd) == false);
-
-	TEST_NNG_PASS(testutil_marry(req, rep));
-
-	// It should be writable now.
-	TEST_CHECK(testutil_pollfd(fd) == true);
-
-	// Submit a bunch of jobs.  Note that we have to stall a bit
-	// between each message to let it queue up.
-	for (int i = 0; i < 10; i++) {
-		int rv = nng_send(req, "", 0, NNG_FLAG_NONBLOCK);
-		if (rv == NNG_EAGAIN) {
-			break;
-		}
-		TEST_NNG_PASS(rv);
-		testutil_sleep(50);
-	}
-	TEST_CHECK(testutil_pollfd(fd) == 0);
-	TEST_NNG_PASS(nng_close(req));
-	TEST_NNG_PASS(nng_close(rep));
-}
-
-void
-test_req_poll_readable(void)
-{
-	int        fd;
-	nng_socket req;
-	nng_socket rep;
-	nng_msg *  msg;
-
-	TEST_NNG_PASS(nng_req0_open(&req));
-	TEST_NNG_PASS(nng_rep0_open(&rep));
-	TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_RECVFD, &fd));
-	TEST_CHECK(fd >= 0);
-
-	// Not readable if not connected!
-	TEST_CHECK(testutil_pollfd(fd) == false);
-
-	// Even after connect (no message yet)
-	TEST_NNG_PASS(testutil_marry(req, rep));
-	TEST_CHECK(testutil_pollfd(fd) == false);
-
-	// But once we send messages, it is.
-	// We have to send a request, in order to send a reply.
-
-	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
-	TEST_NNG_PASS(nng_msg_append(msg, "xyz", 3));
-	TEST_NNG_PASS(nng_sendmsg(req, msg, 0));
-	TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0)); // recv on rep
-	TEST_NNG_PASS(nng_sendmsg(rep, msg, 0));  // echo it back
-	testutil_sleep(200); // give time for message to arrive
-
-	TEST_CHECK(testutil_pollfd(fd) == true);
-
-	// and receiving makes it no longer pollable
-	TEST_NNG_PASS(nng_recvmsg(req, &msg, 0));
-	nng_msg_free(msg);
-	TEST_CHECK(testutil_pollfd(fd) == false);
-
-	// TODO verify unsolicited response
-
-	TEST_NNG_PASS(nng_close(req));
-	TEST_NNG_PASS(nng_close(rep));
-}
-
-void
-test_req_context_not_pollable(void)
-{
-	int        fd;
-	nng_socket req;
-	nng_ctx    ctx;
-
-	TEST_NNG_PASS(nng_req0_open(&req));
-	TEST_NNG_PASS(nng_ctx_open(&ctx, req));
-	TEST_NNG_FAIL(
-	    nng_ctx_getopt_int(ctx, NNG_OPT_SENDFD, &fd), NNG_ENOTSUP);
-	TEST_NNG_FAIL(
-	    nng_ctx_getopt_int(ctx, NNG_OPT_RECVFD, &fd), NNG_ENOTSUP);
-	TEST_NNG_PASS(nng_ctx_close(ctx));
-	TEST_NNG_PASS(nng_close(req));
-}
-
-void
-test_req_validate_peer(void)
-{
-	nng_socket s1, s2;
-	nng_stat * stats;
-	nng_stat * reject;
-	char       addr[64];
-
-	testutil_scratch_addr("inproc", sizeof(addr), addr);
-
-	TEST_NNG_PASS(nng_req0_open(&s1));
-	TEST_NNG_PASS(nng_req0_open(&s2));
-
-	TEST_NNG_PASS(nng_listen(s1, addr, NULL, 0));
-	TEST_NNG_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
-
-	testutil_sleep(100);
-	TEST_NNG_PASS(nng_stats_get(&stats));
-
-	TEST_CHECK(stats != NULL);
-	TEST_CHECK((reject = nng_stat_find_socket(stats, s1)) != NULL);
-	TEST_CHECK((reject = nng_stat_find(reject, "reject")) != NULL);
-
-	TEST_CHECK(nng_stat_type(reject) == NNG_STAT_COUNTER);
-	TEST_CHECK(nng_stat_value(reject) > 0);
-
-	TEST_NNG_PASS(nng_close(s1));
-	TEST_NNG_PASS(nng_close(s2));
-	nng_stats_free(stats);
-}
-
-TEST_LIST = {
-	{ "req rep identity", test_req_rep_identity },
-	{ "resend option", test_resend_option },
-	{ "req recv bad state", test_req_recv_bad_state },
-	{ "rep send bad state", test_rep_send_bad_state },
-	{ "req rep exchange", test_req_rep_exchange },
-	{ "req cancel", test_req_cancel },
-	{ "req cancel abort recv", test_req_cancel_abort_recv },
-	{ "req poll writable", test_req_poll_writeable },
-	{ "req poll readable", test_req_poll_readable },
-	{ "req context not pollable", test_req_context_not_pollable },
-	{ "req validate peer", test_req_validate_peer },
-	{ NULL, NULL },
-};