fixes #1345 Restructure the source tree

This is not quite complete, but it sets the stage for other
protocols (such as zmq or mqtt) to be added to the project.

9 files changed, 4224 insertions, 0 deletions

diff --git a/src/sp/protocol/survey0/CMakeLists.txt b/src/sp/protocol/survey0/CMakeLists.txt
new file mode 100644
index 00000000..b5daca41
--- /dev/null
+++ b/src/sp/protocol/survey0/CMakeLists.txt
@@ -0,0 +1,25 @@
+#
+# 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.
+#
+
+# Surveyor/Respondent protocol
+nng_directory(survey0)
+
+nng_sources_if(NNG_PROTO_SURVEYOR0 survey.c xsurvey.c)
+nng_headers_if(NNG_PROTO_SURVEYOR0 nng/protocol/survey0/survey.h)
+nng_defines_if(NNG_PROTO_SURVEYOR0 NNG_HAVE_SURVEYOR0)
+
+nng_sources_if(NNG_PROTO_RESPONDENT0 respond.c xrespond.c)
+nng_headers_if(NNG_PROTO_RESPONDENT0 nng/protocol/survey0/respond.h)
+nng_defines_if(NNG_PROTO_RESPONDENT0 NNG_HAVE_RESPONDENT0)
+
+nng_test(respond_test)
+nng_test(survey_test)
+nng_test(xrespond_test)
+nng_test(xsurvey_test)
\ No newline at end of file
diff --git a/src/sp/protocol/survey0/respond.c b/src/sp/protocol/survey0/respond.c
new file mode 100644
index 00000000..ad551c8f
--- /dev/null
+++ b/src/sp/protocol/survey0/respond.c
@@ -0,0 +1,693 @@
+//
+// 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 <stdlib.h>
+#include <string.h>
+
+#include "core/nng_impl.h"
+#include "nng/protocol/survey0/respond.h"
+
+// Respondent protocol.  The RESPONDENT protocol is the "replier" side of
+// the surveyor pattern.  This is useful for building service discovery, or
+// voting algorithms, for example.
+
+#ifndef NNI_PROTO_SURVEYOR_V0
+#define NNI_PROTO_SURVEYOR_V0 NNI_PROTO(6, 2)
+#endif
+
+#ifndef NNI_PROTO_RESPONDENT_V0
+#define NNI_PROTO_RESPONDENT_V0 NNI_PROTO(6, 3)
+#endif
+
+typedef struct resp0_pipe resp0_pipe;
+typedef struct resp0_sock resp0_sock;
+typedef struct resp0_ctx  resp0_ctx;
+
+static void resp0_pipe_send_cb(void *);
+static void resp0_pipe_recv_cb(void *);
+static void resp0_pipe_fini(void *);
+
+struct resp0_ctx {
+	resp0_sock *  sock;
+	uint32_t      pipe_id;
+	resp0_pipe *  spipe; // send pipe
+	nni_aio *     saio;  // send aio
+	nni_aio *     raio;  // recv aio
+	nni_list_node sqnode;
+	nni_list_node rqnode;
+	size_t        btrace_len;
+	uint32_t      btrace[NNI_MAX_MAX_TTL + 1];
+};
+
+// resp0_sock is our per-socket protocol private structure.
+struct resp0_sock {
+	nni_mtx        mtx;
+	nni_atomic_int ttl;
+	nni_id_map     pipes;
+	resp0_ctx      ctx;
+	nni_list       recvpipes;
+	nni_list       recvq;
+	nni_pollable   readable;
+	nni_pollable   writable;
+};
+
+// resp0_pipe is our per-pipe protocol private structure.
+struct resp0_pipe {
+	nni_pipe *    npipe;
+	resp0_sock *  psock;
+	bool          busy;
+	bool          closed;
+	uint32_t      id;
+	nni_list      sendq; // contexts waiting to send
+	nni_aio       aio_send;
+	nni_aio       aio_recv;
+	nni_list_node rnode; // receivable linkage
+};
+
+static void
+resp0_ctx_close(void *arg)
+{
+	resp0_ctx * ctx = arg;
+	resp0_sock *s   = ctx->sock;
+	nni_aio *   aio;
+
+	// complete any outstanding operations here, cancellation, etc.
+
+	nni_mtx_lock(&s->mtx);
+	if ((aio = ctx->saio) != NULL) {
+		resp0_pipe *p = ctx->spipe;
+		ctx->saio     = NULL;
+		ctx->spipe    = NULL;
+		nni_list_remove(&p->sendq, ctx);
+		nni_aio_finish_error(aio, NNG_ECLOSED);
+	}
+	if ((aio = ctx->raio) != NULL) {
+		ctx->raio = NULL;
+		nni_list_remove(&s->recvq, ctx);
+		nni_aio_finish_error(aio, NNG_ECLOSED);
+	}
+	nni_mtx_unlock(&s->mtx);
+}
+
+static void
+resp0_ctx_fini(void *arg)
+{
+	resp0_ctx *ctx = arg;
+
+	resp0_ctx_close(ctx);
+}
+
+static int
+resp0_ctx_init(void *carg, void *sarg)
+{
+	resp0_sock *s   = sarg;
+	resp0_ctx * ctx = carg;
+
+	NNI_LIST_NODE_INIT(&ctx->sqnode);
+	NNI_LIST_NODE_INIT(&ctx->rqnode);
+	ctx->btrace_len = 0;
+	ctx->sock       = s;
+	ctx->pipe_id    = 0;
+
+	return (0);
+}
+
+static void
+resp0_ctx_cancel_send(nni_aio *aio, void *arg, int rv)
+{
+	resp0_ctx * ctx = arg;
+	resp0_sock *s   = ctx->sock;
+
+	nni_mtx_lock(&s->mtx);
+	if (ctx->saio != aio) {
+		nni_mtx_unlock(&s->mtx);
+		return;
+	}
+	nni_list_node_remove(&ctx->sqnode);
+	ctx->saio = NULL;
+	nni_mtx_unlock(&s->mtx);
+	nni_msg_header_clear(nni_aio_get_msg(aio)); // reset the headers
+	nni_aio_finish_error(aio, rv);
+}
+
+static void
+resp0_ctx_send(void *arg, nni_aio *aio)
+{
+	resp0_ctx * ctx = arg;
+	resp0_sock *s   = ctx->sock;
+	resp0_pipe *p;
+	nni_msg *   msg;
+	size_t      len;
+	uint32_t    pid;
+	int         rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+	msg = nni_aio_get_msg(aio);
+	nni_msg_header_clear(msg);
+
+	if (ctx == &s->ctx) {
+		// We can't send anymore, because only one send per request.
+		nni_pollable_clear(&s->writable);
+	}
+
+	nni_mtx_lock(&s->mtx);
+	if ((rv = nni_aio_schedule(aio, resp0_ctx_cancel_send, ctx)) != 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	if ((len = ctx->btrace_len) == 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, NNG_ESTATE);
+		return;
+	}
+	pid             = ctx->pipe_id;
+	ctx->pipe_id    = 0;
+	ctx->btrace_len = 0;
+
+	if ((rv = nni_msg_header_append(msg, ctx->btrace, len)) != 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	if ((p = nni_id_get(&s->pipes, pid)) == NULL) {
+		// Surveyor has left the building.  Just discard the reply.
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_set_msg(aio, NULL);
+		nni_aio_finish(aio, 0, nni_msg_len(msg));
+		nni_msg_free(msg);
+		return;
+	}
+
+	if (!p->busy) {
+		p->busy = true;
+		len     = nni_msg_len(msg);
+		nni_aio_set_msg(&p->aio_send, msg);
+		nni_pipe_send(p->npipe, &p->aio_send);
+		nni_mtx_unlock(&s->mtx);
+
+		nni_aio_set_msg(aio, NULL);
+		nni_aio_finish(aio, 0, len);
+		return;
+	}
+
+	ctx->saio  = aio;
+	ctx->spipe = p;
+	nni_list_append(&p->sendq, ctx);
+	nni_mtx_unlock(&s->mtx);
+}
+
+static void
+resp0_sock_fini(void *arg)
+{
+	resp0_sock *s = arg;
+
+	nni_id_map_fini(&s->pipes);
+	resp0_ctx_fini(&s->ctx);
+	nni_pollable_fini(&s->writable);
+	nni_pollable_fini(&s->readable);
+	nni_mtx_fini(&s->mtx);
+}
+
+static int
+resp0_sock_init(void *arg, nni_sock *nsock)
+{
+	resp0_sock *s = arg;
+
+	NNI_ARG_UNUSED(nsock);
+
+	nni_mtx_init(&s->mtx);
+	nni_id_map_init(&s->pipes, 0, 0, false);
+
+	NNI_LIST_INIT(&s->recvq, resp0_ctx, rqnode);
+	NNI_LIST_INIT(&s->recvpipes, resp0_pipe, rnode);
+
+	nni_atomic_init(&s->ttl);
+	nni_atomic_set(&s->ttl, 8); // Per RFC
+
+	(void) resp0_ctx_init(&s->ctx, s);
+
+	// We start off without being either readable or writable.
+	// Readability comes when there is something on the socket.
+	nni_pollable_init(&s->writable);
+	nni_pollable_init(&s->readable);
+	return (0);
+}
+
+static void
+resp0_sock_open(void *arg)
+{
+	NNI_ARG_UNUSED(arg);
+}
+
+static void
+resp0_sock_close(void *arg)
+{
+	resp0_sock *s = arg;
+
+	resp0_ctx_close(&s->ctx);
+}
+
+static void
+resp0_pipe_stop(void *arg)
+{
+	resp0_pipe *p = arg;
+
+	nni_aio_stop(&p->aio_send);
+	nni_aio_stop(&p->aio_recv);
+}
+
+static void
+resp0_pipe_fini(void *arg)
+{
+	resp0_pipe *p = arg;
+	nng_msg *   msg;
+
+	if ((msg = nni_aio_get_msg(&p->aio_recv)) != NULL) {
+		nni_aio_set_msg(&p->aio_recv, NULL);
+		nni_msg_free(msg);
+	}
+	nni_aio_fini(&p->aio_send);
+	nni_aio_fini(&p->aio_recv);
+}
+
+static int
+resp0_pipe_init(void *arg, nni_pipe *npipe, void *s)
+{
+	resp0_pipe *p = arg;
+
+	nni_aio_init(&p->aio_recv, resp0_pipe_recv_cb, p);
+	nni_aio_init(&p->aio_send, resp0_pipe_send_cb, p);
+
+	NNI_LIST_INIT(&p->sendq, resp0_ctx, sqnode);
+
+	p->npipe = npipe;
+	p->psock = s;
+	p->busy  = false;
+	p->id    = nni_pipe_id(npipe);
+
+	return (0);
+}
+
+static int
+resp0_pipe_start(void *arg)
+{
+	resp0_pipe *p = arg;
+	resp0_sock *s = p->psock;
+	int         rv;
+
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_SURVEYOR_V0) {
+		return (NNG_EPROTO);
+	}
+
+	nni_mtx_lock(&s->mtx);
+	rv = nni_id_set(&s->pipes, p->id, p);
+	nni_mtx_unlock(&s->mtx);
+	if (rv != 0) {
+		return (rv);
+	}
+
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+	return (rv);
+}
+
+static void
+resp0_pipe_close(void *arg)
+{
+	resp0_pipe *p = arg;
+	resp0_sock *s = p->psock;
+	resp0_ctx * ctx;
+
+	nni_aio_close(&p->aio_send);
+	nni_aio_close(&p->aio_recv);
+
+	nni_mtx_lock(&s->mtx);
+	p->closed = true;
+	while ((ctx = nni_list_first(&p->sendq)) != NULL) {
+		nni_aio *aio;
+		nni_msg *msg;
+		nni_list_remove(&p->sendq, ctx);
+		aio       = ctx->saio;
+		ctx->saio = NULL;
+		msg       = nni_aio_get_msg(aio);
+		nni_aio_set_msg(aio, NULL);
+		nni_aio_finish(aio, 0, nni_msg_len(msg));
+		nni_msg_free(msg);
+	}
+	if (p->id == s->ctx.pipe_id) {
+		// Make sure user space knows they can send a message to us,
+		// which we will happily discard.
+		nni_pollable_raise(&s->writable);
+	}
+	nni_id_remove(&s->pipes, p->id);
+	nni_mtx_unlock(&s->mtx);
+}
+
+static void
+resp0_pipe_send_cb(void *arg)
+{
+	resp0_pipe *p = arg;
+	resp0_sock *s = p->psock;
+	resp0_ctx * ctx;
+	nni_aio *   aio;
+	nni_msg *   msg;
+	size_t      len;
+
+	if (nni_aio_result(&p->aio_send) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_send));
+		nni_aio_set_msg(&p->aio_send, NULL);
+		nni_pipe_close(p->npipe);
+		return;
+	}
+	nni_mtx_lock(&s->mtx);
+	p->busy = false;
+	if ((ctx = nni_list_first(&p->sendq)) == NULL) {
+		// Nothing else to send.
+		if (p->id == s->ctx.pipe_id) {
+			// Mark us ready for the other side to send!
+			nni_pollable_raise(&s->writable);
+		}
+		nni_mtx_unlock(&s->mtx);
+		return;
+	}
+
+	nni_list_remove(&p->sendq, ctx);
+	aio        = ctx->saio;
+	ctx->saio  = NULL;
+	ctx->spipe = NULL;
+	p->busy    = true;
+	msg        = nni_aio_get_msg(aio);
+	len        = nni_msg_len(msg);
+	nni_aio_set_msg(aio, NULL);
+	nni_aio_set_msg(&p->aio_send, msg);
+	nni_pipe_send(p->npipe, &p->aio_send);
+
+	nni_mtx_unlock(&s->mtx);
+
+	nni_aio_finish_sync(aio, 0, len);
+}
+
+static void
+resp0_cancel_recv(nni_aio *aio, void *arg, int rv)
+{
+	resp0_ctx * ctx = arg;
+	resp0_sock *s   = ctx->sock;
+
+	nni_mtx_lock(&s->mtx);
+	if (ctx->raio == aio) {
+		nni_list_remove(&s->recvq, ctx);
+		ctx->raio = NULL;
+		nni_aio_finish_error(aio, rv);
+	}
+	nni_mtx_unlock(&s->mtx);
+}
+
+static void
+resp0_ctx_recv(void *arg, nni_aio *aio)
+{
+	resp0_ctx * ctx = arg;
+	resp0_sock *s   = ctx->sock;
+	resp0_pipe *p;
+	size_t      len;
+	nni_msg *   msg;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+	nni_mtx_lock(&s->mtx);
+	if ((p = nni_list_first(&s->recvpipes)) == NULL) {
+		int rv;
+		rv = nni_aio_schedule(aio, resp0_cancel_recv, ctx);
+		if (rv != 0) {
+			nni_mtx_unlock(&s->mtx);
+			nni_aio_finish_error(aio, rv);
+			return;
+		}
+		// We cannot have two concurrent receive requests on the same
+		// context...
+		if (ctx->raio != NULL) {
+			nni_mtx_unlock(&s->mtx);
+			nni_aio_finish_error(aio, NNG_ESTATE);
+			return;
+		}
+		ctx->raio = aio;
+		nni_list_append(&s->recvq, ctx);
+		nni_mtx_unlock(&s->mtx);
+		return;
+	}
+	msg = nni_aio_get_msg(&p->aio_recv);
+	nni_aio_set_msg(&p->aio_recv, NULL);
+	nni_list_remove(&s->recvpipes, p);
+	if (nni_list_empty(&s->recvpipes)) {
+		nni_pollable_clear(&s->readable);
+	}
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+
+	len = nni_msg_header_len(msg);
+	memcpy(ctx->btrace, nni_msg_header(msg), len);
+	ctx->btrace_len = len;
+	ctx->pipe_id    = p->id;
+	if (ctx == &s->ctx) {
+		nni_pollable_raise(&s->writable);
+	}
+	nni_mtx_unlock(&s->mtx);
+
+	nni_msg_header_clear(msg);
+	nni_aio_set_msg(aio, msg);
+	nni_aio_finish(aio, 0, nni_msg_len(msg));
+}
+
+static void
+resp0_pipe_recv_cb(void *arg)
+{
+	resp0_pipe *p = arg;
+	resp0_sock *s = p->psock;
+	resp0_ctx * ctx;
+	nni_msg *   msg;
+	nni_aio *   aio;
+	int         hops;
+	size_t      len;
+	int         ttl;
+
+	if (nni_aio_result(&p->aio_recv) != 0) {
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	ttl = nni_atomic_get(&s->ttl);
+	msg = nni_aio_get_msg(&p->aio_recv);
+	nni_msg_set_pipe(msg, p->id);
+
+	// Move backtrace from body to header
+	hops = 1;
+	for (;;) {
+		bool     end = 0;
+		uint8_t *body;
+
+		if (hops > ttl) {
+			goto drop;
+		}
+		hops++;
+		if (nni_msg_len(msg) < 4) {
+			// Peer is speaking garbage, kick it.
+			nni_msg_free(msg);
+			nni_aio_set_msg(&p->aio_recv, NULL);
+			nni_pipe_close(p->npipe);
+			return;
+		}
+		body = nni_msg_body(msg);
+		end  = ((body[0] & 0x80u) != 0);
+		if (nni_msg_header_append(msg, body, 4) != 0) {
+			goto drop;
+		}
+		nni_msg_trim(msg, 4);
+		if (end) {
+			break;
+		}
+	}
+
+	len = nni_msg_header_len(msg);
+
+	nni_mtx_lock(&s->mtx);
+
+	if (p->closed) {
+		// If pipe was closed, we just abandon the data from it.
+		nni_aio_set_msg(&p->aio_recv, NULL);
+		nni_mtx_unlock(&s->mtx);
+		nni_msg_free(msg);
+		return;
+	}
+	if ((ctx = nni_list_first(&s->recvq)) == NULL) {
+		// No one blocked in recv, stall.
+		nni_list_append(&s->recvpipes, p);
+		nni_pollable_raise(&s->readable);
+		nni_mtx_unlock(&s->mtx);
+		return;
+	}
+
+	nni_list_remove(&s->recvq, ctx);
+	aio       = ctx->raio;
+	ctx->raio = NULL;
+	nni_aio_set_msg(&p->aio_recv, NULL);
+
+	// Start the next receive.
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+
+	ctx->btrace_len = len;
+	memcpy(ctx->btrace, nni_msg_header(msg), len);
+	nni_msg_header_clear(msg);
+	ctx->pipe_id = p->id;
+
+	if ((ctx == &s->ctx) && (!p->busy)) {
+		nni_pollable_raise(&s->writable);
+	}
+	nni_mtx_unlock(&s->mtx);
+
+	nni_aio_set_msg(aio, msg);
+	nni_aio_finish_sync(aio, 0, nni_msg_len(msg));
+	return;
+
+drop:
+	nni_msg_free(msg);
+	nni_aio_set_msg(&p->aio_recv, NULL);
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+}
+
+static int
+resp0_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	resp0_sock *s = arg;
+	int         ttl;
+	int         rv;
+
+	if ((rv = nni_copyin_int(&ttl, buf, sz, 1, NNI_MAX_MAX_TTL, t)) == 0) {
+		nni_atomic_set(&s->ttl, ttl);
+	}
+	return (rv);
+}
+
+static int
+resp0_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	resp0_sock *s = arg;
+	return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t));
+}
+
+static int
+resp0_sock_get_sendfd(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	resp0_sock *s = arg;
+	int         rv;
+	int         fd;
+
+	if ((rv = nni_pollable_getfd(&s->writable, &fd)) != 0) {
+		return (rv);
+	}
+	return (nni_copyout_int(fd, buf, szp, t));
+}
+
+static int
+resp0_sock_get_recvfd(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	resp0_sock *s = arg;
+	int         rv;
+	int         fd;
+
+	if ((rv = nni_pollable_getfd(&s->readable, &fd)) != 0) {
+		return (rv);
+	}
+	return (nni_copyout_int(fd, buf, szp, t));
+}
+
+static void
+resp0_sock_send(void *arg, nni_aio *aio)
+{
+	resp0_sock *s = arg;
+
+	resp0_ctx_send(&s->ctx, aio);
+}
+
+static void
+resp0_sock_recv(void *arg, nni_aio *aio)
+{
+	resp0_sock *s = arg;
+
+	resp0_ctx_recv(&s->ctx, aio);
+}
+
+static nni_proto_pipe_ops resp0_pipe_ops = {
+	.pipe_size  = sizeof(resp0_pipe),
+	.pipe_init  = resp0_pipe_init,
+	.pipe_fini  = resp0_pipe_fini,
+	.pipe_start = resp0_pipe_start,
+	.pipe_close = resp0_pipe_close,
+	.pipe_stop  = resp0_pipe_stop,
+};
+
+static nni_proto_ctx_ops resp0_ctx_ops = {
+	.ctx_size = sizeof(resp0_ctx),
+	.ctx_init = resp0_ctx_init,
+	.ctx_fini = resp0_ctx_fini,
+	.ctx_send = resp0_ctx_send,
+	.ctx_recv = resp0_ctx_recv,
+};
+
+static nni_option resp0_sock_options[] = {
+	{
+	    .o_name = NNG_OPT_MAXTTL,
+	    .o_get  = resp0_sock_get_max_ttl,
+	    .o_set  = resp0_sock_set_max_ttl,
+	},
+	{
+	    .o_name = NNG_OPT_RECVFD,
+	    .o_get  = resp0_sock_get_recvfd,
+	    .o_set  = NULL,
+	},
+	{
+	    .o_name = NNG_OPT_SENDFD,
+	    .o_get  = resp0_sock_get_sendfd,
+	    .o_set  = NULL,
+	},
+	// terminate list
+	{
+	    .o_name = NULL,
+	},
+};
+
+static nni_proto_sock_ops resp0_sock_ops = {
+	.sock_size    = sizeof(resp0_sock),
+	.sock_init    = resp0_sock_init,
+	.sock_fini    = resp0_sock_fini,
+	.sock_open    = resp0_sock_open,
+	.sock_close   = resp0_sock_close,
+	.sock_send    = resp0_sock_send,
+	.sock_recv    = resp0_sock_recv,
+	.sock_options = resp0_sock_options,
+};
+
+static nni_proto resp0_proto = {
+	.proto_version  = NNI_PROTOCOL_VERSION,
+	.proto_self     = { NNI_PROTO_RESPONDENT_V0, "respondent" },
+	.proto_peer     = { NNI_PROTO_SURVEYOR_V0, "surveyor" },
+	.proto_flags    = NNI_PROTO_FLAG_SNDRCV,
+	.proto_sock_ops = &resp0_sock_ops,
+	.proto_pipe_ops = &resp0_pipe_ops,
+	.proto_ctx_ops  = &resp0_ctx_ops,
+};
+
+int
+nng_respondent0_open(nng_socket *sidp)
+{
+	return (nni_proto_open(sidp, &resp0_proto));
+}
diff --git a/src/sp/protocol/survey0/respond_test.c b/src/sp/protocol/survey0/respond_test.c
new file mode 100644
index 00000000..51844c76
--- /dev/null
+++ b/src/sp/protocol/survey0/respond_test.c
@@ -0,0 +1,586 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+//
+// 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 <nuts.h>
+
+void
+test_resp_identity(void)
+{
+	nng_socket s;
+	int        p;
+	char *     n;
+
+	NUTS_PASS(nng_respondent0_open(&s));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p));
+	NUTS_TRUE(p == NNG_RESPONDENT0_SELF);
+	NUTS_TRUE(nng_socket_get_int(s, NNG_OPT_PEER, &p) == 0);
+	NUTS_TRUE(p == NNG_RESPONDENT0_PEER);
+	NUTS_TRUE(nng_socket_get_string(s, NNG_OPT_PROTONAME, &n) == 0);
+	NUTS_MATCH(n, NNG_RESPONDENT0_SELF_NAME);
+	nng_strfree(n);
+	NUTS_TRUE(nng_socket_get_string(s, NNG_OPT_PEERNAME, &n) == 0);
+	NUTS_MATCH(n, NNG_RESPONDENT0_PEER_NAME);
+	nng_strfree(n);
+	NUTS_CLOSE(s);
+}
+
+void
+test_resp_send_bad_state(void)
+{
+	nng_socket resp;
+	nng_msg *  msg = NULL;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_msg_alloc(&msg, 0));
+	NUTS_FAIL(nng_sendmsg(resp, msg, 0), NNG_ESTATE);
+	nng_msg_free(msg);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_resp_poll_writeable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_get_int(resp, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Not writable before connect.
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_MARRY(surv, resp);
+
+	// Still not writable.
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// If we get a job, *then* we become writable
+	NUTS_SEND(surv, "abc");
+	NUTS_RECV(resp, "abc");
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	// And is no longer writable once we send a message
+	NUTS_SEND(resp, "def");
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+	// Even after receiving it
+	NUTS_RECV(surv, "def");
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_resp_poll_readable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_get_int(resp, NNG_OPT_RECVFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Not readable if not connected!
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// Even after connect (no message yet)
+	NUTS_MARRY(surv, resp);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// But once we send messages, it is.
+	// We have to send a request, in order to send a reply.
+	NUTS_SEND(surv, "abc");
+	NUTS_SLEEP(100);
+
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	// and receiving makes it no longer ready
+	NUTS_PASS(nng_recvmsg(resp, &msg, 0));
+	nng_msg_free(msg);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// TODO verify unsolicited response
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_resp_context_no_poll(void)
+{
+	int        fd;
+	nng_socket resp;
+	nng_ctx    ctx;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_ctx_open(&ctx, resp));
+	NUTS_FAIL(nng_ctx_get_int(ctx, NNG_OPT_SENDFD, &fd), NNG_ENOTSUP);
+	NUTS_FAIL(nng_ctx_get_int(ctx, NNG_OPT_RECVFD, &fd), NNG_ENOTSUP);
+	NUTS_PASS(nng_ctx_close(ctx));
+	NUTS_CLOSE(resp);
+}
+
+void
+test_resp_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char *     addr;
+
+	NUTS_ADDR(addr, "inproc");
+
+	NUTS_PASS(nng_respondent0_open(&s1));
+	NUTS_PASS(nng_respondent0_open(&s2));
+
+	NUTS_PASS(nng_listen(s1, addr, NULL, 0));
+	NUTS_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	NUTS_SLEEP(100);
+	NUTS_PASS(nng_stats_get(&stats));
+
+	NUTS_TRUE(stats != NULL);
+	NUTS_TRUE((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	NUTS_TRUE((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	NUTS_TRUE(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	NUTS_TRUE(nng_stat_value(reject) > 0);
+
+	NUTS_CLOSE(s1);
+	NUTS_CLOSE(s2);
+	nng_stats_free(stats);
+}
+
+void
+test_resp_double_recv(void)
+{
+	nng_socket s1;
+	nng_aio *  aio1;
+	nng_aio *  aio2;
+
+	NUTS_PASS(nng_respondent0_open(&s1));
+	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
+	NUTS_PASS(nng_aio_alloc(&aio2, NULL, NULL));
+
+	nng_recv_aio(s1, aio1);
+	nng_recv_aio(s1, aio2);
+
+	nng_aio_wait(aio2);
+	NUTS_FAIL(nng_aio_result(aio2), NNG_ESTATE);
+	NUTS_CLOSE(s1);
+	NUTS_FAIL(nng_aio_result(aio1), NNG_ECLOSED);
+	nng_aio_free(aio1);
+	nng_aio_free(aio2);
+}
+
+void
+test_resp_close_pipe_before_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_pipe   p;
+	nng_aio *  aio1;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
+
+	NUTS_MARRY(surv, resp);
+	NUTS_SEND(surv, "test");
+
+	nng_recv_aio(resp, aio1);
+	nng_aio_wait(aio1);
+	NUTS_PASS(nng_aio_result(aio1));
+	NUTS_TRUE((m = nng_aio_get_msg(aio1)) != NULL);
+	p = nng_msg_get_pipe(m);
+	NUTS_PASS(nng_pipe_close(p));
+	NUTS_PASS(nng_sendmsg(resp, m, 0));
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio1);
+}
+
+void
+test_resp_close_pipe_during_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_pipe   p = NNG_PIPE_INITIALIZER;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_SENDTIMEO, 200));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_SENDBUF, 20));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_RECVBUF, 20));
+	NUTS_PASS(nng_setopt_int(surv, NNG_OPT_SENDBUF, 20));
+	NUTS_PASS(nng_setopt_int(surv, NNG_OPT_RECVBUF, 1));
+
+	NUTS_MARRY(surv, resp);
+
+	for (int i = 0; i < 100; i++) {
+		int rv;
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_append_u32(m, (unsigned) i | 0x80000000u));
+		NUTS_PASS(nng_sendmsg(surv, m, 0));
+		NUTS_PASS(nng_recvmsg(resp, &m, 0));
+		p  = nng_msg_get_pipe(m);
+		rv = nng_sendmsg(resp, m, 0);
+		if (rv == NNG_ETIMEDOUT) {
+			// Queue is backed up, senders are busy.
+			nng_msg_free(m);
+			break;
+		}
+		NUTS_PASS(rv);
+	}
+	NUTS_PASS(nng_pipe_close(p));
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_resp_ctx_recv_aio_stopped(void)
+{
+	nng_socket resp;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_ctx_open(&ctx, resp));
+
+	nng_aio_stop(aio);
+	nng_ctx_recv(ctx, aio);
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_PASS(nng_ctx_close(ctx));
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio);
+}
+
+void
+test_resp_close_pipe_context_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_pipe   p = NNG_PIPE_INITIALIZER;
+	nng_msg *  m;
+	nng_ctx    ctx[10];
+	nng_aio *  aio[10];
+	int        i;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_SENDBUF, 1));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_RECVBUF, 1));
+	NUTS_PASS(nng_setopt_int(surv, NNG_OPT_SENDBUF, 1));
+	NUTS_PASS(nng_setopt_int(surv, NNG_OPT_RECVBUF, 1));
+	for (i = 0; i < 10; i++) {
+		NUTS_PASS(nng_ctx_open(&ctx[i], resp));
+		NUTS_PASS(nng_aio_alloc(&aio[i], NULL, NULL));
+	}
+
+	NUTS_MARRY(surv, resp);
+
+	for (i = 0; i < 10; i++) {
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_append_u32(m, (unsigned) i | 0x80000000u));
+		NUTS_PASS(nng_sendmsg(surv, m, 0));
+		nng_ctx_recv(ctx[i], aio[i]);
+	}
+	for (i = 0; i < 10; i++) {
+		nng_aio_wait(aio[i]);
+		NUTS_PASS(nng_aio_result(aio[i]));
+		NUTS_TRUE((m = nng_aio_get_msg(aio[i])) != NULL);
+		p = nng_msg_get_pipe(m);
+		nng_aio_set_msg(aio[i], m);
+		nng_ctx_send(ctx[i], aio[i]);
+	}
+
+	// Note that SURVEYOR socket is not reading the results.
+	NUTS_PASS(nng_pipe_close(p));
+
+	for (i = 0; i < 10; i++) {
+		int rv;
+		nng_aio_wait(aio[i]);
+		rv = nng_aio_result(aio[i]);
+		if (rv != 0) {
+			NUTS_FAIL(rv, NNG_ECLOSED);
+			nng_msg_free(nng_aio_get_msg(aio[i]));
+		}
+		nng_aio_free(aio[i]);
+		NUTS_PASS(nng_ctx_close(ctx[i]));
+	}
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_resp_close_context_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+	nng_ctx    ctx[10];
+	nng_aio *  aio[10];
+	int        i;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_SENDBUF, 1));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_RECVBUF, 1));
+	NUTS_PASS(nng_setopt_int(surv, NNG_OPT_SENDBUF, 1));
+	NUTS_PASS(nng_setopt_int(surv, NNG_OPT_RECVBUF, 1));
+	for (i = 0; i < 10; i++) {
+		NUTS_PASS(nng_ctx_open(&ctx[i], resp));
+		NUTS_PASS(nng_aio_alloc(&aio[i], NULL, NULL));
+	}
+
+	NUTS_MARRY(surv, resp);
+
+	for (i = 0; i < 10; i++) {
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_append_u32(m, (unsigned) i | 0x80000000u));
+		NUTS_PASS(nng_sendmsg(surv, m, 0));
+		nng_ctx_recv(ctx[i], aio[i]);
+	}
+	for (i = 0; i < 10; i++) {
+		nng_aio_wait(aio[i]);
+		NUTS_PASS(nng_aio_result(aio[i]));
+		NUTS_TRUE((m = nng_aio_get_msg(aio[i])) != NULL);
+		nng_aio_set_msg(aio[i], m);
+		nng_ctx_send(ctx[i], aio[i]);
+	}
+
+	// Note that REQ socket is not reading the results.
+	for (i = 0; i < 10; i++) {
+		int rv;
+		NUTS_PASS(nng_ctx_close(ctx[i]));
+		nng_aio_wait(aio[i]);
+		rv = nng_aio_result(aio[i]);
+		if (rv != 0) {
+			NUTS_FAIL(rv, NNG_ECLOSED);
+			nng_msg_free(nng_aio_get_msg(aio[i]));
+		}
+		nng_aio_free(aio[i]);
+	}
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_resp_ctx_recv_nonblock(void)
+{
+	nng_socket resp;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_ctx_open(&ctx, resp));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_ctx_recv(ctx, aio);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio);
+}
+
+static void
+test_resp_ctx_send_nonblock(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_ctx_open(&ctx, resp));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_MARRY(surv, resp);
+
+	NUTS_SEND(surv, "SEND");
+	nng_ctx_recv(ctx, aio);
+	nng_aio_wait(aio);
+	NUTS_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);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio);
+	nng_msg_free(msg);
+}
+
+void
+test_resp_recv_garbage(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 200));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_SENDTIMEO, 200));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_PASS(nng_msg_alloc(&m, 4));
+	NUTS_PASS(nng_msg_append_u32(m, 1u));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+	NUTS_FAIL(nng_recvmsg(resp, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_resp_ttl_option(void)
+{
+	nng_socket  resp;
+	int         v;
+	bool        b;
+	size_t      sz;
+	const char *opt = NNG_OPT_MAXTTL;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+
+	NUTS_PASS(nng_setopt_int(resp, opt, 1));
+	NUTS_FAIL(nng_setopt_int(resp, opt, 0), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_int(resp, opt, -1), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_int(resp, opt, 16), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_int(resp, opt, 256), NNG_EINVAL);
+	NUTS_PASS(nng_setopt_int(resp, opt, 3));
+	NUTS_PASS(nng_socket_get_int(resp, opt, &v));
+	NUTS_TRUE(v == 3);
+	v  = 0;
+	sz = sizeof(v);
+	NUTS_PASS(nng_socket_get(resp, opt, &v, &sz));
+	NUTS_TRUE(v == 3);
+	NUTS_TRUE(sz == sizeof(v));
+
+	NUTS_FAIL(nng_setopt(resp, opt, "", 1), NNG_EINVAL);
+	sz = 1;
+	NUTS_FAIL(nng_socket_get(resp, opt, &v, &sz), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_bool(resp, opt, true), NNG_EBADTYPE);
+	NUTS_FAIL(nng_socket_get_bool(resp, opt, &b), NNG_EBADTYPE);
+
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_resp_ttl_drop(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_setopt_int(resp, NNG_OPT_MAXTTL, 3));
+	NUTS_PASS(nng_setopt_ms(resp, NNG_OPT_RECVTIMEO, 200));
+	NUTS_PASS(nng_setopt_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	// Send messages.  Note that xrep implicitly adds a hop on receive.
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 2 hops
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000001u));
+	NUTS_PASS(nng_msg_append(m, "PASS1", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 4 hops -- discard!
+	NUTS_PASS(nng_msg_append_u32(m, 2u));
+	NUTS_PASS(nng_msg_append_u32(m, 3u));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000002u));
+	NUTS_PASS(nng_msg_append(m, "FAIL2", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 3 hops - passes
+	NUTS_PASS(nng_msg_append_u32(m, 2u));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000003u));
+	NUTS_PASS(nng_msg_append(m, "PASS3", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 4 hops -- discard!
+	NUTS_PASS(nng_msg_append_u32(m, 2u));
+	NUTS_PASS(nng_msg_append_u32(m, 3u));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000003u));
+	NUTS_PASS(nng_msg_append(m, "FAIL4", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_RECV(resp, "PASS1");
+	NUTS_RECV(resp, "PASS3");
+
+	NUTS_FAIL(nng_recvmsg(resp, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(resp);
+	NUTS_CLOSE(surv);
+}
+
+TEST_LIST = {
+	{ "respond identity", test_resp_identity },
+	{ "respond send bad state", test_resp_send_bad_state },
+	{ "respond poll readable", test_resp_poll_readable },
+	{ "respond poll writable", test_resp_poll_writeable },
+	{ "respond context does not poll", test_resp_context_no_poll },
+	{ "respond validate peer", test_resp_validate_peer },
+	{ "respond double recv", test_resp_double_recv },
+	{ "respond close pipe before send", test_resp_close_pipe_before_send },
+	{ "respond close pipe during send", test_resp_close_pipe_during_send },
+	{ "respond recv aio ctx stopped", test_resp_ctx_recv_aio_stopped },
+	{ "respond close pipe context send",
+	    test_resp_close_pipe_context_send },
+	{ "respond close context send", test_resp_close_context_send },
+	{ "respond context send nonblock", test_resp_ctx_send_nonblock },
+	{ "respond context recv nonblock", test_resp_ctx_recv_nonblock },
+	{ "respond recv garbage", test_resp_recv_garbage },
+	{ "respond ttl option", test_resp_ttl_option },
+	{ "respond ttl drop", test_resp_ttl_drop },
+	{ NULL, NULL },
+};
diff --git a/src/sp/protocol/survey0/survey.c b/src/sp/protocol/survey0/survey.c
new file mode 100644
index 00000000..ce1ed601
--- /dev/null
+++ b/src/sp/protocol/survey0/survey.c
@@ -0,0 +1,663 @@
+//
+// 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 <stdlib.h>
+
+#include "core/nng_impl.h"
+#include "nng/protocol/survey0/survey.h"
+
+// Surveyor protocol.  The SURVEYOR protocol is the "survey" side of the
+// survey pattern.  This is useful for building service discovery, voting, etc.
+// Note that this pattern is not optimized for extreme low latency, as it makes
+// multiple use of queues for simplicity.  Typically this is used in cases
+// where a few dozen extra microseconds does not matter.
+
+typedef struct surv0_pipe surv0_pipe;
+typedef struct surv0_sock surv0_sock;
+typedef struct surv0_ctx  surv0_ctx;
+
+static void surv0_pipe_send_cb(void *);
+static void surv0_pipe_recv_cb(void *);
+static void surv0_ctx_timeout(void *);
+
+struct surv0_ctx {
+	surv0_sock *   sock;
+	uint32_t       survey_id; // survey id
+	nni_timer_node timer;
+	nni_time       expire;
+	nni_lmq        recv_lmq;
+	nni_list       recv_queue;
+	nni_atomic_int recv_buf;
+	nni_atomic_int survey_time;
+	int            err;
+};
+
+// surv0_sock is our per-socket protocol private structure.
+struct surv0_sock {
+	int            ttl;
+	nni_list       pipes;
+	nni_mtx        mtx;
+	surv0_ctx      ctx;
+	nni_id_map     surveys;
+	nni_pollable   writable;
+	nni_pollable   readable;
+	nni_atomic_int send_buf;
+};
+
+// surv0_pipe is our per-pipe protocol private structure.
+struct surv0_pipe {
+	nni_pipe *    pipe;
+	surv0_sock *  sock;
+	nni_lmq       send_queue;
+	nni_list_node node;
+	nni_aio       aio_send;
+	nni_aio       aio_recv;
+	bool          busy;
+	bool          closed;
+};
+
+static void
+surv0_ctx_abort(surv0_ctx *ctx, int err)
+{
+	nni_aio *   aio;
+	surv0_sock *sock = ctx->sock;
+
+	while ((aio = nni_list_first(&ctx->recv_queue)) != NULL) {
+		nni_list_remove(&ctx->recv_queue, aio);
+		nni_aio_finish_error(aio, err);
+	}
+	nni_lmq_flush(&ctx->recv_lmq);
+	if (ctx->survey_id != 0) {
+		nni_id_remove(&sock->surveys, ctx->survey_id);
+		ctx->survey_id = 0;
+	}
+	if (ctx == &sock->ctx) {
+		nni_pollable_clear(&sock->readable);
+	}
+}
+
+static void
+surv0_ctx_close(surv0_ctx *ctx)
+{
+	surv0_sock *sock = ctx->sock;
+
+	nni_mtx_lock(&sock->mtx);
+	surv0_ctx_abort(ctx, NNG_ECLOSED);
+	nni_mtx_unlock(&sock->mtx);
+}
+
+static void
+surv0_ctx_fini(void *arg)
+{
+	surv0_ctx *ctx = arg;
+
+	surv0_ctx_close(ctx);
+	nni_timer_cancel(&ctx->timer);
+	nni_lmq_fini(&ctx->recv_lmq);
+}
+
+static int
+surv0_ctx_init(void *c, void *s)
+{
+	surv0_ctx *  ctx  = c;
+	surv0_sock * sock = s;
+	int          rv;
+	int          len;
+	nng_duration tmo;
+
+	nni_aio_list_init(&ctx->recv_queue);
+	nni_atomic_init(&ctx->recv_buf);
+	nni_atomic_init(&ctx->survey_time);
+
+	if (ctx == &sock->ctx) {
+		len = 128;
+		tmo = NNI_SECOND; // survey timeout
+	} else {
+		len = nni_atomic_get(&sock->ctx.recv_buf);
+		tmo = nni_atomic_get(&sock->ctx.survey_time);
+	}
+
+	nni_atomic_set(&ctx->recv_buf, len);
+	nni_atomic_set(&ctx->survey_time, tmo);
+
+	ctx->sock = sock;
+
+	if ((rv = nni_lmq_init(&ctx->recv_lmq, len)) != 0) {
+		surv0_ctx_fini(ctx);
+		return (rv);
+	}
+	nni_timer_init(&ctx->timer, surv0_ctx_timeout, ctx);
+	return (0);
+}
+
+static void
+surv0_ctx_cancel(nni_aio *aio, void *arg, int rv)
+{
+	surv0_ctx * ctx  = arg;
+	surv0_sock *sock = ctx->sock;
+	nni_mtx_lock(&sock->mtx);
+	if (nni_list_active(&ctx->recv_queue, aio)) {
+		nni_list_remove(&ctx->recv_queue, aio);
+		nni_aio_finish_error(aio, rv);
+	}
+	if (ctx->survey_id != 0) {
+		nni_id_remove(&sock->surveys, ctx->survey_id);
+		ctx->survey_id = 0;
+	}
+	nni_mtx_unlock(&sock->mtx);
+}
+
+static void
+surv0_ctx_recv(void *arg, nni_aio *aio)
+{
+	surv0_ctx * ctx  = arg;
+	surv0_sock *sock = ctx->sock;
+	nni_msg *   msg;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+
+	nni_mtx_lock(&sock->mtx);
+	if (ctx->survey_id == 0) {
+		nni_mtx_unlock(&sock->mtx);
+		nni_aio_finish_error(aio, NNG_ESTATE);
+		return;
+	}
+again:
+	if (nni_lmq_getq(&ctx->recv_lmq, &msg) != 0) {
+		int rv;
+		if ((rv = nni_aio_schedule(aio, &surv0_ctx_cancel, ctx)) !=
+		    0) {
+			nni_mtx_unlock(&sock->mtx);
+			nni_aio_finish_error(aio, rv);
+			return;
+		}
+		nni_list_append(&ctx->recv_queue, aio);
+		nni_mtx_unlock(&sock->mtx);
+		return;
+	}
+	if (nni_lmq_empty(&ctx->recv_lmq) && (ctx == &sock->ctx)) {
+		nni_pollable_clear(&sock->readable);
+	}
+	if ((msg = nni_msg_unique(msg)) == NULL) {
+		goto again;
+	}
+
+	nni_mtx_unlock(&sock->mtx);
+	nni_aio_finish_msg(aio, msg);
+}
+
+void
+surv0_ctx_timeout(void *arg)
+{
+	surv0_ctx * ctx  = arg;
+	surv0_sock *sock = ctx->sock;
+
+	nni_mtx_lock(&sock->mtx);
+	if (nni_clock() < ctx->expire) {
+		nni_mtx_unlock(&sock->mtx);
+		return;
+	}
+
+	// Abort any pending receives.
+	surv0_ctx_abort(ctx, NNG_ETIMEDOUT);
+	nni_mtx_unlock(&sock->mtx);
+}
+
+static void
+surv0_ctx_send(void *arg, nni_aio *aio)
+{
+	surv0_ctx *  ctx  = arg;
+	surv0_sock * sock = ctx->sock;
+	surv0_pipe * pipe;
+	nni_msg *    msg = nni_aio_get_msg(aio);
+	size_t       len = nni_msg_len(msg);
+	nni_time     now = nni_clock();
+	nng_duration survey_time;
+	int          rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+
+	survey_time = nni_atomic_get(&ctx->survey_time);
+
+	nni_mtx_lock(&sock->mtx);
+
+	// Abort everything outstanding.
+	surv0_ctx_abort(ctx, NNG_ECANCELED);
+	nni_timer_cancel(&ctx->timer);
+
+	// Allocate the new ID.
+	if ((rv = nni_id_alloc(&sock->surveys, &ctx->survey_id, ctx)) != 0) {
+		nni_mtx_unlock(&sock->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	nni_msg_header_clear(msg);
+	nni_msg_header_append_u32(msg, (uint32_t) ctx->survey_id);
+
+	// From this point, we're committed to success.  Note that we send
+	// regardless of whether there are any pipes or not.  If no pipes,
+	// then it just gets discarded.
+	nni_aio_set_msg(aio, NULL);
+	NNI_LIST_FOREACH (&sock->pipes, pipe) {
+
+		// if the pipe isn't busy, then send this message direct.
+		if (!pipe->busy) {
+			pipe->busy = true;
+			nni_msg_clone(msg);
+			nni_aio_set_msg(&pipe->aio_send, msg);
+			nni_pipe_send(pipe->pipe, &pipe->aio_send);
+		} else if (!nni_lmq_full(&pipe->send_queue)) {
+			nni_msg_clone(msg);
+			nni_lmq_putq(&pipe->send_queue, msg);
+		}
+	}
+
+	ctx->expire = now + survey_time;
+	nni_timer_schedule(&ctx->timer, ctx->expire);
+
+	nni_mtx_unlock(&sock->mtx);
+	nni_msg_free(msg);
+
+	nni_aio_finish(aio, 0, len);
+}
+
+static void
+surv0_sock_fini(void *arg)
+{
+	surv0_sock *sock = arg;
+
+	surv0_ctx_fini(&sock->ctx);
+	nni_id_map_fini(&sock->surveys);
+	nni_pollable_fini(&sock->writable);
+	nni_pollable_fini(&sock->readable);
+	nni_mtx_fini(&sock->mtx);
+}
+
+static int
+surv0_sock_init(void *arg, nni_sock *s)
+{
+	surv0_sock *sock = arg;
+	int         rv;
+
+	NNI_ARG_UNUSED(s);
+
+	NNI_LIST_INIT(&sock->pipes, surv0_pipe, node);
+	nni_mtx_init(&sock->mtx);
+	nni_pollable_init(&sock->readable);
+	nni_pollable_init(&sock->writable);
+	// We are always writable.
+	nni_pollable_raise(&sock->writable);
+
+	// We allow for some buffering on a per pipe basis, to allow for
+	// multiple contexts to have surveys outstanding.  It is recommended
+	// to increase this if many contexts will want to publish
+	// at nearly the same time.
+	nni_atomic_init(&sock->send_buf);
+	nni_atomic_set(&sock->send_buf, 8);
+
+	// Survey IDs are 32 bits, with the high order bit set.
+	// We start at a random point, to minimize likelihood of
+	// accidental collision across restarts.
+	nni_id_map_init(&sock->surveys, 0x80000000u, 0xffffffffu, true);
+
+	if ((rv = surv0_ctx_init(&sock->ctx, sock)) != 0) {
+		surv0_sock_fini(sock);
+		return (rv);
+	}
+
+	sock->ttl = 8;
+
+	return (0);
+}
+
+static void
+surv0_sock_open(void *arg)
+{
+	NNI_ARG_UNUSED(arg);
+}
+
+static void
+surv0_sock_close(void *arg)
+{
+	surv0_sock *s = arg;
+
+	surv0_ctx_close(&s->ctx);
+}
+
+static void
+surv0_pipe_stop(void *arg)
+{
+	surv0_pipe *p = arg;
+
+	nni_aio_stop(&p->aio_send);
+	nni_aio_stop(&p->aio_recv);
+}
+
+static void
+surv0_pipe_fini(void *arg)
+{
+	surv0_pipe *p = arg;
+
+	nni_aio_fini(&p->aio_send);
+	nni_aio_fini(&p->aio_recv);
+	nni_lmq_fini(&p->send_queue);
+}
+
+static int
+surv0_pipe_init(void *arg, nni_pipe *pipe, void *s)
+{
+	surv0_pipe *p    = arg;
+	surv0_sock *sock = s;
+	int         rv;
+	int         len;
+
+	len = nni_atomic_get(&sock->send_buf);
+	nni_aio_init(&p->aio_send, surv0_pipe_send_cb, p);
+	nni_aio_init(&p->aio_recv, surv0_pipe_recv_cb, p);
+
+	// This depth could be tunable.  The deeper the queue, the more
+	// concurrent surveys that can be delivered (multiple contexts).
+	// Note that surveys can be *outstanding*, but not yet put on the wire.
+	if ((rv = nni_lmq_init(&p->send_queue, len)) != 0) {
+		surv0_pipe_fini(p);
+		return (rv);
+	}
+
+	p->pipe = pipe;
+	p->sock = sock;
+	return (0);
+}
+
+static int
+surv0_pipe_start(void *arg)
+{
+	surv0_pipe *p = arg;
+	surv0_sock *s = p->sock;
+
+	if (nni_pipe_peer(p->pipe) != NNG_SURVEYOR0_PEER) {
+		return (NNG_EPROTO);
+	}
+
+	nni_mtx_lock(&s->mtx);
+	nni_list_append(&s->pipes, p);
+	nni_mtx_unlock(&s->mtx);
+
+	nni_pipe_recv(p->pipe, &p->aio_recv);
+	return (0);
+}
+
+static void
+surv0_pipe_close(void *arg)
+{
+	surv0_pipe *p = arg;
+	surv0_sock *s = p->sock;
+
+	nni_aio_close(&p->aio_send);
+	nni_aio_close(&p->aio_recv);
+
+	nni_mtx_lock(&s->mtx);
+	p->closed = true;
+	nni_lmq_flush(&p->send_queue);
+	if (nni_list_active(&s->pipes, p)) {
+		nni_list_remove(&s->pipes, p);
+	}
+	nni_mtx_unlock(&s->mtx);
+}
+
+static void
+surv0_pipe_send_cb(void *arg)
+{
+	surv0_pipe *p    = arg;
+	surv0_sock *sock = p->sock;
+	nni_msg *   msg;
+
+	if (nni_aio_result(&p->aio_send) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_send));
+		nni_aio_set_msg(&p->aio_send, NULL);
+		nni_pipe_close(p->pipe);
+		return;
+	}
+
+	nni_mtx_lock(&sock->mtx);
+	if (p->closed) {
+		nni_mtx_unlock(&sock->mtx);
+		return;
+	}
+	if (nni_lmq_getq(&p->send_queue, &msg) == 0) {
+		nni_aio_set_msg(&p->aio_send, msg);
+		nni_pipe_send(p->pipe, &p->aio_send);
+	} else {
+		p->busy = false;
+	}
+	nni_mtx_unlock(&sock->mtx);
+}
+
+static void
+surv0_pipe_recv_cb(void *arg)
+{
+	surv0_pipe *p    = arg;
+	surv0_sock *sock = p->sock;
+	surv0_ctx * ctx;
+	nni_msg *   msg;
+	uint32_t    id;
+	nni_aio *   aio;
+
+	if (nni_aio_result(&p->aio_recv) != 0) {
+		nni_pipe_close(p->pipe);
+		return;
+	}
+
+	msg = nni_aio_get_msg(&p->aio_recv);
+	nni_aio_set_msg(&p->aio_recv, NULL);
+	nni_msg_set_pipe(msg, nni_pipe_id(p->pipe));
+
+	// We yank 4 bytes of body, and move them to the header.
+	if (nni_msg_len(msg) < 4) {
+		// Peer sent us garbage.  Kick it.
+		nni_msg_free(msg);
+		nni_pipe_close(p->pipe);
+		return;
+	}
+	id = nni_msg_trim_u32(msg);
+	nni_msg_header_append_u32(msg, id);
+
+	nni_mtx_lock(&sock->mtx);
+	// Best effort at delivery.  Discard if no context or context is
+	// unable to receive it.
+	if (((ctx = nni_id_get(&sock->surveys, id)) == NULL) ||
+	    (nni_lmq_full(&ctx->recv_lmq))) {
+		nni_msg_free(msg);
+	} else if ((aio = nni_list_first(&ctx->recv_queue)) != NULL) {
+		nni_list_remove(&ctx->recv_queue, aio);
+		nni_aio_finish_msg(aio, msg);
+	} else {
+		nni_lmq_putq(&ctx->recv_lmq, msg);
+		if (ctx == &sock->ctx) {
+			nni_pollable_raise(&sock->readable);
+		}
+	}
+	nni_mtx_unlock(&sock->mtx);
+
+	nni_pipe_recv(p->pipe, &p->aio_recv);
+}
+
+static int
+surv0_ctx_set_survey_time(
+    void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	surv0_ctx *  ctx = arg;
+	nng_duration expire;
+	int          rv;
+	if ((rv = nni_copyin_ms(&expire, buf, sz, t)) == 0) {
+		nni_atomic_set(&ctx->survey_time, expire);
+	}
+	return (rv);
+}
+
+static int
+surv0_ctx_get_survey_time(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	surv0_ctx *ctx = arg;
+	return (
+	    nni_copyout_ms(nni_atomic_get(&ctx->survey_time), buf, szp, t));
+}
+
+static int
+surv0_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	surv0_sock *s = arg;
+	return (nni_copyin_int(&s->ttl, buf, sz, 1, NNI_MAX_MAX_TTL, t));
+}
+
+static int
+surv0_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	surv0_sock *s = arg;
+	return (nni_copyout_int(s->ttl, buf, szp, t));
+}
+
+static int
+surv0_sock_set_survey_time(
+    void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	surv0_sock *s = arg;
+	return (surv0_ctx_set_survey_time(&s->ctx, buf, sz, t));
+}
+
+static int
+surv0_sock_get_survey_time(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	surv0_sock *s = arg;
+	return (surv0_ctx_get_survey_time(&s->ctx, buf, szp, t));
+}
+
+static int
+surv0_sock_get_send_fd(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	surv0_sock *sock = arg;
+	int         rv;
+	int         fd;
+
+	if ((rv = nni_pollable_getfd(&sock->writable, &fd)) != 0) {
+		return (rv);
+	}
+	return (nni_copyout_int(fd, buf, szp, t));
+}
+
+static int
+surv0_sock_get_recv_fd(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	surv0_sock *sock = arg;
+	int         rv;
+	int         fd;
+
+	if ((rv = nni_pollable_getfd(&sock->readable, &fd)) != 0) {
+		return (rv);
+	}
+	return (nni_copyout_int(fd, buf, szp, t));
+}
+
+static void
+surv0_sock_recv(void *arg, nni_aio *aio)
+{
+	surv0_sock *s = arg;
+	surv0_ctx_recv(&s->ctx, aio);
+}
+
+static void
+surv0_sock_send(void *arg, nni_aio *aio)
+{
+	surv0_sock *s = arg;
+	surv0_ctx_send(&s->ctx, aio);
+}
+
+static nni_proto_pipe_ops surv0_pipe_ops = {
+	.pipe_size  = sizeof(surv0_pipe),
+	.pipe_init  = surv0_pipe_init,
+	.pipe_fini  = surv0_pipe_fini,
+	.pipe_start = surv0_pipe_start,
+	.pipe_close = surv0_pipe_close,
+	.pipe_stop  = surv0_pipe_stop,
+};
+
+static nni_option surv0_ctx_options[] = {
+	{
+	    .o_name = NNG_OPT_SURVEYOR_SURVEYTIME,
+	    .o_get  = surv0_ctx_get_survey_time,
+	    .o_set  = surv0_ctx_set_survey_time,
+	},
+	{
+	    .o_name = NULL,
+	}
+};
+static nni_proto_ctx_ops surv0_ctx_ops = {
+	.ctx_size    = sizeof(surv0_ctx),
+	.ctx_init    = surv0_ctx_init,
+	.ctx_fini    = surv0_ctx_fini,
+	.ctx_send    = surv0_ctx_send,
+	.ctx_recv    = surv0_ctx_recv,
+	.ctx_options = surv0_ctx_options,
+};
+
+static nni_option surv0_sock_options[] = {
+	{
+	    .o_name = NNG_OPT_SURVEYOR_SURVEYTIME,
+	    .o_get  = surv0_sock_get_survey_time,
+	    .o_set  = surv0_sock_set_survey_time,
+	},
+	{
+	    .o_name = NNG_OPT_MAXTTL,
+	    .o_get  = surv0_sock_get_max_ttl,
+	    .o_set  = surv0_sock_set_max_ttl,
+	},
+	{
+	    .o_name = NNG_OPT_RECVFD,
+	    .o_get  = surv0_sock_get_recv_fd,
+	},
+	{
+	    .o_name = NNG_OPT_SENDFD,
+	    .o_get  = surv0_sock_get_send_fd,
+	},
+	// terminate list
+	{
+	    .o_name = NULL,
+	},
+};
+
+static nni_proto_sock_ops surv0_sock_ops = {
+	.sock_size    = sizeof(surv0_sock),
+	.sock_init    = surv0_sock_init,
+	.sock_fini    = surv0_sock_fini,
+	.sock_open    = surv0_sock_open,
+	.sock_close   = surv0_sock_close,
+	.sock_send    = surv0_sock_send,
+	.sock_recv    = surv0_sock_recv,
+	.sock_options = surv0_sock_options,
+};
+
+static nni_proto surv0_proto = {
+	.proto_version  = NNI_PROTOCOL_VERSION,
+	.proto_self     = { NNG_SURVEYOR0_SELF, NNG_SURVEYOR0_SELF_NAME },
+	.proto_peer     = { NNG_SURVEYOR0_PEER, NNG_SURVEYOR0_PEER_NAME },
+	.proto_flags    = NNI_PROTO_FLAG_SNDRCV,
+	.proto_sock_ops = &surv0_sock_ops,
+	.proto_pipe_ops = &surv0_pipe_ops,
+	.proto_ctx_ops  = &surv0_ctx_ops,
+};
+
+int
+nng_surveyor0_open(nng_socket *sock)
+{
+	return (nni_proto_open(sock, &surv0_proto));
+}
diff --git a/src/sp/protocol/survey0/survey_test.c b/src/sp/protocol/survey0/survey_test.c
new file mode 100644
index 00000000..95d27adf
--- /dev/null
+++ b/src/sp/protocol/survey0/survey_test.c
@@ -0,0 +1,626 @@
+//
+// 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 <nuts.h>
+
+static void
+test_surv_identity(void)
+{
+	nng_socket s;
+	int        p;
+	char *     n;
+
+	NUTS_PASS(nng_surveyor0_open(&s));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p));
+	NUTS_TRUE(p == NNG_SURVEYOR0_SELF);
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PEER, &p));
+	NUTS_TRUE(p == NNG_SURVEYOR0_PEER); // 49
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PROTONAME, &n));
+	NUTS_MATCH(n, NNG_SURVEYOR0_SELF_NAME);
+	nng_strfree(n);
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PEERNAME, &n));
+	NUTS_MATCH(n, NNG_SURVEYOR0_PEER_NAME);
+	nng_strfree(n);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_surv_ttl_option(void)
+{
+	nng_socket  surv;
+	int         v;
+	bool        b;
+	size_t      sz;
+	const char *opt = NNG_OPT_MAXTTL;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+
+	NUTS_PASS(nng_socket_set_int(surv, opt, 1));
+	NUTS_FAIL(nng_socket_set_int(surv, opt, 0), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(surv, opt, -1), NNG_EINVAL);
+	// This test will fail if the NNI_MAX_MAX_TTL is changed from the
+	// builtin default of 15.
+	NUTS_FAIL(nng_socket_set_int(surv, opt, 16), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(surv, opt, 256), NNG_EINVAL);
+	NUTS_PASS(nng_socket_set_int(surv, opt, 3));
+	NUTS_PASS(nng_socket_get_int(surv, opt, &v));
+	NUTS_TRUE(v == 3);
+	v  = 0;
+	sz = sizeof(v);
+	NUTS_PASS(nng_socket_get(surv, opt, &v, &sz));
+	NUTS_TRUE(v == 3);
+	NUTS_TRUE(sz == sizeof(v));
+
+	NUTS_FAIL(nng_socket_set(surv, opt, "", 1), NNG_EINVAL);
+	sz = 1;
+	NUTS_FAIL(nng_socket_get(surv, opt, &v, &sz), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_bool(surv, opt, true), NNG_EBADTYPE);
+	NUTS_FAIL(nng_socket_get_bool(surv, opt, &b), NNG_EBADTYPE);
+
+	NUTS_CLOSE(surv);
+}
+
+static void
+test_surv_survey_time_option(void)
+{
+	nng_socket   surv;
+	nng_duration d;
+	bool         b;
+	size_t       sz  = sizeof(b);
+	const char * opt = NNG_OPT_SURVEYOR_SURVEYTIME;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+
+	NUTS_PASS(nng_socket_set_ms(surv, opt, 10));
+	NUTS_FAIL(nng_socket_set(surv, opt, "", 1), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_get(surv, opt, &b, &sz), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_bool(surv, opt, true), NNG_EBADTYPE);
+	NUTS_FAIL(nng_socket_get_bool(surv, opt, &b), NNG_EBADTYPE);
+
+	NUTS_PASS(nng_socket_get_ms(surv, opt, &d));
+	NUTS_TRUE(d == 10);
+	NUTS_CLOSE(surv);
+}
+
+void
+test_surv_recv_bad_state(void)
+{
+	nng_socket surv;
+	nng_msg *  msg = NULL;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_FAIL(nng_recvmsg(surv, &msg, 0), NNG_ESTATE);
+	NUTS_TRUE(msg == NULL);
+	NUTS_CLOSE(surv);
+}
+
+static void
+test_surv_recv_garbage(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+	uint32_t   surv_id;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_recvmsg(resp, &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.
+	NUTS_TRUE(nng_msg_header_len(m) == 8);
+	NUTS_PASS(nng_msg_header_chop_u32(m, &surv_id));
+
+	NUTS_PASS(nng_sendmsg(resp, m, 0));
+	NUTS_FAIL(nng_recvmsg(surv, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+#define SECOND 1000
+
+void
+test_surv_resp_exchange(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, SECOND));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, SECOND));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, SECOND));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, SECOND));
+
+	NUTS_MARRY(resp, surv);
+
+	NUTS_SEND(surv, "ping");
+	NUTS_RECV(resp, "ping");
+	NUTS_SEND(resp, "pong");
+	NUTS_RECV(surv, "pong");
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_surv_cancel(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open(&surv));
+
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, SECOND));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, SECOND));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 5 * SECOND));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 5 * SECOND));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_SENDBUF, 16));
+
+	NUTS_MARRY(resp, surv);
+
+	// Send req #1 (abc).
+	NUTS_SEND(surv, "abc");
+
+	// 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.)
+	NUTS_SLEEP(100);
+
+	// Send the next next request ("def").  Note that
+	// the RESP side server will have already buffered the receive
+	// request, and should simply be waiting for us to reply to abc.
+	NUTS_SEND(surv, "def");
+
+	// Receive the first request (should be abc) on the REP server.
+	NUTS_RECV(resp, "abc");
+
+	// RESP sends the reply to first command.  This will be discarded
+	// by the SURV socket.
+	NUTS_SEND(resp, "abc");
+
+	// Now get the next command from the REP; should be "def".
+	NUTS_RECV(resp, "def");
+
+	// And send it back to REQ.
+	NUTS_SEND(resp, "def");
+
+	// Try a req command.  This should give back "def"
+	NUTS_RECV(surv, "def");
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_surv_cancel_abort_recv(void)
+{
+	nng_aio *    aio;
+	nng_duration time = SECOND * 10; // 10s (kind of never)
+	nng_socket   surv;
+	nng_socket   resp;
+
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SURVEYOR_SURVEYTIME, time));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_SENDBUF, 16));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 5 * SECOND));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 5 * SECOND));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 5 * SECOND));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 5 * SECOND));
+
+	NUTS_MARRY(resp, surv);
+
+	// Send survey #1 (abc).
+	NUTS_SEND(surv, "abc");
+
+	// Wait for it to get ot the other side.
+	NUTS_SLEEP(100);
+
+	nng_aio_set_timeout(aio, 5 * SECOND);
+	nng_recv_aio(surv, aio);
+
+	// Give time for this recv to post properly.
+	NUTS_SLEEP(100);
+
+	// Send the next next request ("def").  Note that
+	// the respondent side server will have already buffered the receive
+	// request, and should simply be waiting for us to reply to
+	// abc.
+	NUTS_SEND(surv, "def");
+
+	// Our pending I/O should have been canceled.
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+
+	// Receive the first request (should be abc) on the respondent.
+	NUTS_RECV(resp, "abc");
+
+	// Respondent sends the reply to first survey.  This will be
+	// discarded by the SURV socket.
+	NUTS_SEND(resp, "abc");
+
+	// Now get the next survey from the RESP; should be "def".
+	NUTS_RECV(resp, "def");
+
+	// And send it back to REQ.
+	NUTS_SEND(resp, "def");
+
+	// Try a req command.  This should give back "def"
+	NUTS_RECV(surv, "def");
+
+	nng_aio_free(aio);
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_surv_cancel_post_recv(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_MARRY(surv, resp);
+
+	NUTS_SEND(surv, "ONE");
+	NUTS_RECV(resp, "ONE");
+	NUTS_SEND(resp, "one");
+	NUTS_SLEEP(100); // Make sure reply arrives!
+	NUTS_SEND(surv, "TWO");
+	NUTS_RECV(resp, "TWO");
+	NUTS_SEND(resp, "two");
+	NUTS_RECV(surv, "two");
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_surv_poll_writeable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_get_int(surv, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Survey is broadcast, so we can always write.
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	NUTS_MARRY(surv, resp);
+
+	// Now it's writable.
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+void
+test_surv_poll_readable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_get_int(surv, NNG_OPT_RECVFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Not readable if not connected!
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// Even after connect (no message yet)
+	NUTS_MARRY(surv, resp);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// But once we send messages, it is.
+	// We have to send a request, in order to send a reply.
+
+	NUTS_PASS(nng_msg_alloc(&msg, 0));
+	NUTS_PASS(nng_msg_append(msg, "xyz", 3));
+	NUTS_PASS(nng_sendmsg(surv, msg, 0));
+	NUTS_PASS(nng_recvmsg(resp, &msg, 0)); // recv on rep
+	NUTS_PASS(nng_sendmsg(resp, msg, 0));  // echo it back
+	NUTS_SLEEP(200); // give time for message to arrive
+
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	// and receiving makes it no longer ready
+	NUTS_PASS(nng_recvmsg(surv, &msg, 0));
+	nng_msg_free(msg);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// TODO verify unsolicited response
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_surv_ctx_no_poll(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_ctx    ctx;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_ctx_open(&ctx, surv));
+	NUTS_FAIL(nng_ctx_get_int(ctx, NNG_OPT_SENDFD, &fd), NNG_ENOTSUP);
+	NUTS_FAIL(nng_ctx_get_int(ctx, NNG_OPT_RECVFD, &fd), NNG_ENOTSUP);
+	NUTS_PASS(nng_ctx_close(ctx));
+	NUTS_CLOSE(surv);
+}
+
+static void
+test_surv_ctx_recv_nonblock(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_ctx_open(&ctx, surv));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&msg, 0));
+
+	NUTS_MARRY(surv, resp);
+
+	nng_aio_set_msg(aio, msg);
+	nng_ctx_send(ctx, aio);
+	nng_aio_wait(aio);
+	NUTS_PASS(nng_aio_result(aio));
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_ctx_recv(ctx, aio);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio);
+}
+
+static void
+test_surv_ctx_send_nonblock(void)
+{
+	nng_socket surv;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_ctx_open(&ctx, surv));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_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);
+	NUTS_PASS(nng_aio_result(aio)); // We never block
+	NUTS_CLOSE(surv);
+	nng_aio_free(aio);
+}
+
+static void
+test_surv_send_best_effort(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_MARRY(surv, resp);
+
+	for (int i = 0; i < 200; i++) {
+		NUTS_SEND(surv, "junk");
+	}
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_surv_survey_timeout(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+	char       buf[16];
+	size_t     sz;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SURVEYOR_SURVEYTIME, 50));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 100));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_SEND(surv, "hello");
+	NUTS_RECV(resp, "hello");
+
+	sz = sizeof(buf);
+	NUTS_FAIL(nng_recv(surv, buf, &sz, 0), NNG_ETIMEDOUT);
+	NUTS_SEND(resp, "world");
+	NUTS_FAIL(nng_recv(surv, buf, &sz, 0), NNG_ESTATE);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_surv_ctx_recv_close_socket(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+	nng_ctx    ctx;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_ctx_open(&ctx, surv));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_MARRY(surv, resp);
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	nng_aio_set_msg(aio, m);
+	nng_ctx_send(ctx, aio);
+	nng_aio_wait(aio);
+	NUTS_PASS(nng_aio_result(aio));
+
+	nng_ctx_recv(ctx, aio);
+	nng_close(surv);
+
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECLOSED);
+	nng_aio_free(aio);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_surv_context_multi(void)
+{
+	nng_socket surv;
+	nng_socket resp;
+	nng_ctx    c[5];
+	nng_aio *  aio;
+	nng_msg *  m;
+	int        cnt = sizeof(c) / sizeof(c[0]);
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_MARRY(surv, resp);
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SURVEYOR_SURVEYTIME, 200));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	for (int i = 0; i < cnt; i++) {
+		NUTS_PASS(nng_ctx_open(&c[i], surv));
+	}
+
+	for (int i = 0; i < cnt; i++) {
+		NUTS_PASS(nng_msg_alloc(&m, 0));
+		NUTS_PASS(nng_msg_append_u32(m, i));
+		nng_aio_set_msg(aio, m);
+		nng_ctx_send(c[i], aio);
+		nng_aio_wait(aio);
+		NUTS_PASS(nng_aio_result(aio));
+	}
+
+	for (int i = 0; i < cnt; i++) {
+		NUTS_PASS(nng_recvmsg(resp, &m, 0));
+		NUTS_PASS(nng_sendmsg(resp, m, 0));
+	}
+
+	for (int i = cnt - 1; i >= 0; i--) {
+		uint32_t x;
+		nng_ctx_recv(c[i], aio);
+		nng_aio_wait(aio);
+		NUTS_PASS(nng_aio_result(aio));
+		m = nng_aio_get_msg(aio);
+		TEST_ASSERT(m != NULL);
+		NUTS_PASS(nng_msg_trim_u32(m, &x));
+		NUTS_TRUE(x == (uint32_t) i);
+		nng_msg_free(m);
+	}
+
+	for (int i = 0; i < cnt; i++) {
+		nng_ctx_recv(c[i], aio);
+		nng_aio_wait(aio);
+		NUTS_TRUE(nng_aio_result(aio) != 0);
+	}
+	for (int i = 0; i < cnt; i++) {
+		nng_ctx_close(c[i]);
+	}
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio);
+}
+
+static void
+test_surv_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char *     addr;
+
+	NUTS_ADDR(addr, "inproc");
+	NUTS_PASS(nng_surveyor0_open(&s1));
+	NUTS_PASS(nng_surveyor0_open(&s2));
+
+	NUTS_PASS(nng_listen(s1, addr, NULL, 0));
+	NUTS_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	NUTS_SLEEP(100);
+	NUTS_PASS(nng_stats_get(&stats));
+
+	NUTS_TRUE(stats != NULL);
+	NUTS_TRUE((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	NUTS_TRUE((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	NUTS_TRUE(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	NUTS_TRUE(nng_stat_value(reject) > 0);
+
+	NUTS_PASS(nng_close(s1));
+	NUTS_PASS(nng_close(s2));
+	nng_stats_free(stats);
+}
+
+TEST_LIST = {
+	{ "survey identity", test_surv_identity },
+	{ "survey ttl option", test_surv_ttl_option },
+	{ "survey survey time option", test_surv_survey_time_option },
+	{ "survey recv bad state", test_surv_recv_bad_state },
+	{ "survey recv garbage", test_surv_recv_garbage },
+	{ "survey respondent exchange", test_surv_resp_exchange },
+	{ "survey cancel", test_surv_cancel },
+	{ "survey cancel abort recv", test_surv_cancel_abort_recv },
+	{ "survey cancel post recv", test_surv_cancel_post_recv },
+	{ "survey poll writable", test_surv_poll_writeable },
+	{ "survey poll readable", test_surv_poll_readable },
+	{ "survey context does not poll", test_surv_ctx_no_poll },
+	{ "survey context recv close socket",
+	    test_surv_ctx_recv_close_socket },
+	{ "survey context recv nonblock", test_surv_ctx_recv_nonblock },
+	{ "survey context send nonblock", test_surv_ctx_send_nonblock },
+	{ "survey timeout", test_surv_survey_timeout },
+	{ "survey send best effort", test_surv_send_best_effort },
+	{ "survey context multi", test_surv_context_multi },
+	{ "survey validate peer", test_surv_validate_peer },
+	{ NULL, NULL },
+};
diff --git a/src/sp/protocol/survey0/xrespond.c b/src/sp/protocol/survey0/xrespond.c
new file mode 100644
index 00000000..b2f203c3
--- /dev/null
+++ b/src/sp/protocol/survey0/xrespond.c
@@ -0,0 +1,417 @@
+//
+// 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 <stdlib.h>
+
+#include "core/nng_impl.h"
+#include "nng/protocol/survey0/respond.h"
+
+// Respondent protocol.  The RESPONDENT protocol is the "replier" side of
+// the surveyor pattern.  This is useful for building service discovery, or
+// voting algorithms, for example.
+
+#ifndef NNI_PROTO_SURVEYOR_V0
+#define NNI_PROTO_SURVEYOR_V0 NNI_PROTO(6, 2)
+#endif
+
+#ifndef NNI_PROTO_RESPONDENT_V0
+#define NNI_PROTO_RESPONDENT_V0 NNI_PROTO(6, 3)
+#endif
+
+typedef struct xresp0_pipe xresp0_pipe;
+typedef struct xresp0_sock xresp0_sock;
+
+static void xresp0_recv_cb(void *);
+static void xresp0_putq_cb(void *);
+static void xresp0_getq_cb(void *);
+static void xresp0_send_cb(void *);
+static void xresp0_sock_getq_cb(void *);
+static void xresp0_pipe_fini(void *);
+
+// resp0_sock is our per-socket protocol private structure.
+struct xresp0_sock {
+	nni_msgq *     urq;
+	nni_msgq *     uwq;
+	nni_atomic_int ttl;
+	nni_id_map     pipes;
+	nni_aio        aio_getq;
+	nni_mtx        mtx;
+};
+
+// resp0_pipe is our per-pipe protocol private structure.
+struct xresp0_pipe {
+	nni_pipe *   npipe;
+	xresp0_sock *psock;
+	uint32_t     id;
+	nni_msgq *   sendq;
+	nni_aio      aio_getq;
+	nni_aio      aio_putq;
+	nni_aio      aio_send;
+	nni_aio      aio_recv;
+};
+
+static void
+xresp0_sock_fini(void *arg)
+{
+	xresp0_sock *s = arg;
+
+	nni_aio_fini(&s->aio_getq);
+	nni_id_map_fini(&s->pipes);
+	nni_mtx_fini(&s->mtx);
+}
+
+static int
+xresp0_sock_init(void *arg, nni_sock *nsock)
+{
+	xresp0_sock *s = arg;
+
+	nni_mtx_init(&s->mtx);
+	nni_atomic_init(&s->ttl);
+	nni_atomic_set(&s->ttl, 8); // Per RFC
+	nni_id_map_init(&s->pipes, 0, 0, false);
+	nni_aio_init(&s->aio_getq, xresp0_sock_getq_cb, s);
+
+	s->urq = nni_sock_recvq(nsock);
+	s->uwq = nni_sock_sendq(nsock);
+
+	return (0);
+}
+
+static void
+xresp0_sock_open(void *arg)
+{
+	xresp0_sock *s = arg;
+
+	nni_msgq_aio_get(s->uwq, &s->aio_getq);
+}
+
+static void
+xresp0_sock_close(void *arg)
+{
+	xresp0_sock *s = arg;
+
+	nni_aio_close(&s->aio_getq);
+}
+
+static void
+xresp0_pipe_stop(void *arg)
+{
+	xresp0_pipe *p = arg;
+
+	nni_aio_stop(&p->aio_putq);
+	nni_aio_stop(&p->aio_getq);
+	nni_aio_stop(&p->aio_send);
+	nni_aio_stop(&p->aio_recv);
+}
+
+static void
+xresp0_pipe_fini(void *arg)
+{
+	xresp0_pipe *p = arg;
+
+	nni_aio_fini(&p->aio_putq);
+	nni_aio_fini(&p->aio_getq);
+	nni_aio_fini(&p->aio_send);
+	nni_aio_fini(&p->aio_recv);
+	nni_msgq_fini(p->sendq);
+}
+
+static int
+xresp0_pipe_init(void *arg, nni_pipe *npipe, void *s)
+{
+	xresp0_pipe *p = arg;
+	int          rv;
+
+	nni_aio_init(&p->aio_putq, xresp0_putq_cb, p);
+	nni_aio_init(&p->aio_recv, xresp0_recv_cb, p);
+	nni_aio_init(&p->aio_getq, xresp0_getq_cb, p);
+	nni_aio_init(&p->aio_send, xresp0_send_cb, p);
+
+	if ((rv = nni_msgq_init(&p->sendq, 2)) != 0) {
+		xresp0_pipe_fini(p);
+		return (rv);
+	}
+
+	p->npipe = npipe;
+	p->psock = s;
+	return (0);
+}
+
+static int
+xresp0_pipe_start(void *arg)
+{
+	xresp0_pipe *p = arg;
+	xresp0_sock *s = p->psock;
+	int          rv;
+
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_SURVEYOR_V0) {
+		return (NNG_EPROTO);
+	}
+
+	p->id = nni_pipe_id(p->npipe);
+
+	nni_mtx_lock(&s->mtx);
+	rv = nni_id_set(&s->pipes, p->id, p);
+	nni_mtx_unlock(&s->mtx);
+	if (rv != 0) {
+		return (rv);
+	}
+
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+	nni_msgq_aio_get(p->sendq, &p->aio_getq);
+
+	return (rv);
+}
+
+static void
+xresp0_pipe_close(void *arg)
+{
+	xresp0_pipe *p = arg;
+	xresp0_sock *s = p->psock;
+
+	nni_aio_close(&p->aio_putq);
+	nni_aio_close(&p->aio_getq);
+	nni_aio_close(&p->aio_send);
+	nni_aio_close(&p->aio_recv);
+
+	nni_msgq_close(p->sendq);
+
+	nni_mtx_lock(&s->mtx);
+	nni_id_remove(&s->pipes, p->id);
+	nni_mtx_unlock(&s->mtx);
+}
+
+// resp0_sock_send watches for messages from the upper write queue,
+// extracts the destination pipe, and forwards it to the appropriate
+// destination pipe via a separate queue.  This prevents a single bad
+// or slow pipe from gumming up the works for the entire socket.s
+
+void
+xresp0_sock_getq_cb(void *arg)
+{
+	xresp0_sock *s = arg;
+	nni_msg *    msg;
+	uint32_t     id;
+	xresp0_pipe *p;
+
+	if (nni_aio_result(&s->aio_getq) != 0) {
+		return;
+	}
+	msg = nni_aio_get_msg(&s->aio_getq);
+	nni_aio_set_msg(&s->aio_getq, NULL);
+
+	// We yank the outgoing pipe id from the header
+	if (nni_msg_header_len(msg) < 4) {
+		nni_msg_free(msg);
+		// We can't really close down the socket, so just keep going.
+		nni_msgq_aio_get(s->uwq, &s->aio_getq);
+		return;
+	}
+	id = nni_msg_header_trim_u32(msg);
+
+	nni_mtx_lock(&s->mtx);
+	// Look for the pipe, and attempt to put the message there
+	// (nonblocking) if we can.  If we can't for any reason, then we
+	// free the message.
+	if (((p = nni_id_get(&s->pipes, id)) == NULL) ||
+	    (nni_msgq_tryput(p->sendq, msg) != 0)) {
+		nni_msg_free(msg);
+	}
+	nni_mtx_unlock(&s->mtx);
+	nni_msgq_aio_get(s->uwq, &s->aio_getq);
+}
+
+void
+xresp0_getq_cb(void *arg)
+{
+	xresp0_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_getq) != 0) {
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	nni_aio_set_msg(&p->aio_send, nni_aio_get_msg(&p->aio_getq));
+	nni_aio_set_msg(&p->aio_getq, NULL);
+
+	nni_pipe_send(p->npipe, &p->aio_send);
+}
+
+void
+xresp0_send_cb(void *arg)
+{
+	xresp0_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_send) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_send));
+		nni_aio_set_msg(&p->aio_send, NULL);
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	nni_msgq_aio_get(p->sendq, &p->aio_getq);
+}
+
+static void
+xresp0_recv_cb(void *arg)
+{
+	xresp0_pipe *p   = arg;
+	xresp0_sock *s   = p->psock;
+	nni_msgq *   urq = s->urq;
+	nni_msg *    msg;
+	int          hops;
+	int          ttl;
+
+	if (nni_aio_result(&p->aio_recv) != 0) {
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	ttl = nni_atomic_get(&s->ttl);
+	msg = nni_aio_get_msg(&p->aio_recv);
+	nni_aio_set_msg(&p->aio_recv, NULL);
+	nni_msg_set_pipe(msg, p->id);
+
+	// Store the pipe id in the header, first thing.
+	nni_msg_header_append_u32(msg, p->id);
+
+	// Move backtrace from body to header
+	hops = 1;
+	for (;;) {
+		bool     end;
+		uint8_t *body;
+
+		if (hops > ttl) {
+			goto drop;
+		}
+		hops++;
+		if (nni_msg_len(msg) < 4) {
+			// Peer sent us garbage, so kick it.
+			nni_msg_free(msg);
+			nni_pipe_close(p->npipe);
+			return;
+		}
+		body = nni_msg_body(msg);
+		end  = ((body[0] & 0x80u) != 0);
+		if (nni_msg_header_append(msg, body, 4) != 0) {
+			goto drop;
+		}
+		nni_msg_trim(msg, 4);
+		if (end) {
+			break;
+		}
+	}
+
+	// Now send it up.
+	nni_aio_set_msg(&p->aio_putq, msg);
+	nni_msgq_aio_put(urq, &p->aio_putq);
+	return;
+
+drop:
+	nni_msg_free(msg);
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+}
+
+static void
+xresp0_putq_cb(void *arg)
+{
+	xresp0_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_putq) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_putq));
+		nni_aio_set_msg(&p->aio_putq, NULL);
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+}
+
+static int
+xresp0_sock_set_maxttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	xresp0_sock *s = arg;
+	int          ttl;
+	int          rv;
+	if ((rv = nni_copyin_int(&ttl, buf, sz, 1, NNI_MAX_MAX_TTL, t)) == 0) {
+		nni_atomic_set(&s->ttl, ttl);
+	}
+	return (rv);
+}
+
+static int
+xresp0_sock_get_maxttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	xresp0_sock *s = arg;
+	return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t));
+}
+
+static void
+xresp0_sock_send(void *arg, nni_aio *aio)
+{
+	xresp0_sock *s = arg;
+
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+xresp0_sock_recv(void *arg, nni_aio *aio)
+{
+	xresp0_sock *s = arg;
+
+	nni_msgq_aio_get(s->urq, aio);
+}
+
+static nni_proto_pipe_ops xresp0_pipe_ops = {
+	.pipe_size  = sizeof(xresp0_pipe),
+	.pipe_init  = xresp0_pipe_init,
+	.pipe_fini  = xresp0_pipe_fini,
+	.pipe_start = xresp0_pipe_start,
+	.pipe_close = xresp0_pipe_close,
+	.pipe_stop  = xresp0_pipe_stop,
+};
+
+static nni_option xresp0_sock_options[] = {
+	{
+	    .o_name = NNG_OPT_MAXTTL,
+	    .o_get  = xresp0_sock_get_maxttl,
+	    .o_set  = xresp0_sock_set_maxttl,
+	},
+	// terminate list
+	{
+	    .o_name = NULL,
+	},
+};
+
+static nni_proto_sock_ops xresp0_sock_ops = {
+	.sock_size    = sizeof(xresp0_sock),
+	.sock_init    = xresp0_sock_init,
+	.sock_fini    = xresp0_sock_fini,
+	.sock_open    = xresp0_sock_open,
+	.sock_close   = xresp0_sock_close,
+	.sock_send    = xresp0_sock_send,
+	.sock_recv    = xresp0_sock_recv,
+	.sock_options = xresp0_sock_options,
+};
+
+static nni_proto xresp0_proto = {
+	.proto_version  = NNI_PROTOCOL_VERSION,
+	.proto_self     = { NNI_PROTO_RESPONDENT_V0, "respondent" },
+	.proto_peer     = { NNI_PROTO_SURVEYOR_V0, "surveyor" },
+	.proto_flags    = NNI_PROTO_FLAG_SNDRCV | NNI_PROTO_FLAG_RAW,
+	.proto_sock_ops = &xresp0_sock_ops,
+	.proto_pipe_ops = &xresp0_pipe_ops,
+};
+
+int
+nng_respondent0_open_raw(nng_socket *sidp)
+{
+	return (nni_proto_open(sidp, &xresp0_proto));
+}
diff --git a/src/sp/protocol/survey0/xrespond_test.c b/src/sp/protocol/survey0/xrespond_test.c
new file mode 100644
index 00000000..ec5e99a3
--- /dev/null
+++ b/src/sp/protocol/survey0/xrespond_test.c
@@ -0,0 +1,436 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+//
+// 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 <nuts.h>
+
+static void
+test_xresp_identity(void)
+{
+	nng_socket s;
+	int        p1, p2;
+	char *     n1;
+	char *     n2;
+
+	NUTS_PASS(nng_respondent0_open_raw(&s));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p1));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PEER, &p2));
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PROTONAME, &n1));
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PEERNAME, &n2));
+	NUTS_CLOSE(s);
+	NUTS_TRUE(p1 == NNG_RESPONDENT0_SELF);
+	NUTS_TRUE(p2 == NNG_RESPONDENT0_PEER);
+	NUTS_MATCH(n1, NNG_RESPONDENT0_SELF_NAME);
+	NUTS_MATCH(n2, NNG_RESPONDENT0_PEER_NAME);
+	nng_strfree(n1);
+	nng_strfree(n2);
+}
+
+static void
+test_xresp_raw(void)
+{
+	nng_socket s;
+	bool       b;
+
+	NUTS_PASS(nng_respondent0_open_raw(&s));
+	NUTS_PASS(nng_socket_get_bool(s, NNG_OPT_RAW, &b));
+	NUTS_TRUE(b);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_xresp_no_context(void)
+{
+	nng_socket s;
+	nng_ctx    ctx;
+
+	NUTS_PASS(nng_respondent0_open_raw(&s));
+	NUTS_FAIL(nng_ctx_open(&ctx, s), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_xresp_poll_writeable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_socket_get_int(resp, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// We are always writeable, even before connect.  This is so that
+	// back-pressure from a bad peer can't trash others.  We assume
+	// that peers won't send us requests faster than they can consume
+	// the answers.  If they do, they will lose their answers.
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	NUTS_MARRY(surv, resp);
+
+	// Now it's writable.
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_poll_readable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_socket_get_int(resp, NNG_OPT_RECVFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Not readable if not connected!
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// Even after connect (no message yet)
+	NUTS_MARRY(surv, resp);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// But once we send messages, it is.
+	// We have to send a request, in order to send a reply.
+	NUTS_SEND(surv, "abc");
+	NUTS_SLEEP(100);
+
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	// and receiving makes it no longer ready
+	NUTS_PASS(nng_recvmsg(resp, &msg, 0));
+	nng_msg_free(msg);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char *     addr;
+
+	NUTS_ADDR(addr, "inproc");
+
+	NUTS_PASS(nng_respondent0_open_raw(&s1));
+	NUTS_PASS(nng_respondent0_open(&s2));
+
+	NUTS_PASS(nng_listen(s1, addr, NULL, 0));
+	NUTS_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	NUTS_SLEEP(100);
+	NUTS_PASS(nng_stats_get(&stats));
+
+	NUTS_TRUE(stats != NULL);
+	NUTS_TRUE((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	NUTS_TRUE((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	NUTS_TRUE(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	NUTS_TRUE(nng_stat_value(reject) > 0);
+
+	NUTS_CLOSE(s1);
+	NUTS_CLOSE(s2);
+	nng_stats_free(stats);
+}
+
+static void
+test_xresp_close_pipe_before_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_pipe   p;
+	nng_aio *  aio1;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open(&surv));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
+
+	NUTS_MARRY(surv, resp);
+	NUTS_SEND(surv, "test");
+
+	nng_recv_aio(resp, aio1);
+	nng_aio_wait(aio1);
+	NUTS_PASS(nng_aio_result(aio1));
+	NUTS_TRUE((m = nng_aio_get_msg(aio1)) != NULL);
+	p = nng_msg_get_pipe(m);
+	NUTS_PASS(nng_pipe_close(p));
+	NUTS_PASS(nng_sendmsg(resp, m, 0));
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio1);
+}
+
+static void
+test_xresp_close_pipe_during_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_pipe   p;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 200));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_int(resp, NNG_OPT_SENDBUF, 20));
+	NUTS_PASS(nng_socket_set_int(resp, NNG_OPT_RECVBUF, 20));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_SENDBUF, 20));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_RECVBUF, 1));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_PASS(nng_msg_alloc(&m, 4));
+	NUTS_PASS(nng_msg_append_u32(m, (unsigned) 0x81000000u));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+	NUTS_PASS(nng_recvmsg(resp, &m, 0));
+	p = nng_msg_get_pipe(m);
+	nng_msg_free(m);
+
+	for (int i = 0; i < 100; i++) {
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_header_append_u32(m, nng_pipe_id(p)));
+		NUTS_PASS(
+		    nng_msg_header_append_u32(m, (unsigned) i | 0x80000000u));
+		// protocol does not exert back-pressure
+		NUTS_PASS(nng_sendmsg(resp, m, 0));
+	}
+	NUTS_PASS(nng_pipe_close(p));
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_close_during_recv(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 100));
+	NUTS_PASS(nng_socket_set_int(resp, NNG_OPT_RECVBUF, 5));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_SENDBUF, 20));
+
+	NUTS_MARRY(surv, resp);
+
+	for (unsigned i = 0; i < 100; i++) {
+		int rv;
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_header_append_u32(m, i | 0x80000000u));
+		rv = nng_sendmsg(surv, m, 0);
+		if (rv == NNG_ETIMEDOUT) {
+			nng_msg_free(m);
+			break;
+		}
+	}
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_recv_aio_stopped(void)
+{
+	nng_socket resp;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_stop(aio);
+	nng_recv_aio(resp, aio);
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_CLOSE(resp);
+	nng_aio_free(aio);
+}
+
+static void
+test_xresp_send_no_header(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_PASS(nng_msg_alloc(&m, 4));
+	NUTS_PASS(nng_sendmsg(resp, m, 0));
+	NUTS_FAIL(nng_recvmsg(resp, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_recv_garbage(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_PASS(nng_msg_alloc(&m, 4));
+	NUTS_PASS(nng_msg_append_u32(m, 1u));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+	NUTS_FAIL(nng_recvmsg(resp, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_ttl_option(void)
+{
+	nng_socket  resp;
+	int         v;
+	bool        b;
+	size_t      sz;
+	const char *opt = NNG_OPT_MAXTTL;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+
+	NUTS_PASS(nng_socket_set_int(resp, opt, 1));
+	NUTS_FAIL(nng_socket_set_int(resp, opt, 0), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(resp, opt, -1), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(resp, opt, 16), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(resp, opt, 256), NNG_EINVAL);
+	NUTS_PASS(nng_socket_set_int(resp, opt, 3));
+	NUTS_PASS(nng_socket_get_int(resp, opt, &v));
+	NUTS_TRUE(v == 3);
+	v  = 0;
+	sz = sizeof(v);
+	NUTS_PASS(nng_socket_get(resp, opt, &v, &sz));
+	NUTS_TRUE(v == 3);
+	NUTS_TRUE(sz == sizeof(v));
+
+	NUTS_FAIL(nng_socket_set(resp, opt, "", 1), NNG_EINVAL);
+	sz = 1;
+	NUTS_FAIL(nng_socket_get(resp, opt, &v, &sz), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_bool(resp, opt, true), NNG_EBADTYPE);
+	NUTS_FAIL(nng_socket_get_bool(resp, opt, &b), NNG_EBADTYPE);
+
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xresp_ttl_drop(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_int(resp, NNG_OPT_MAXTTL, 3));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 200));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	// Send messages.  Note that xresp implicitly adds a hop on receive.
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 2 hops
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000001u));
+	NUTS_PASS(nng_msg_append(m, "PASS1", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 4 hops -- discard!
+	NUTS_PASS(nng_msg_append_u32(m, 2u));
+	NUTS_PASS(nng_msg_append_u32(m, 3u));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000002u));
+	NUTS_PASS(nng_msg_append(m, "FAIL2", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 3 hops - passes
+	NUTS_PASS(nng_msg_append_u32(m, 2u));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000003u));
+	NUTS_PASS(nng_msg_append(m, "PASS3", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 1u)); // 4 hops -- discard!
+	NUTS_PASS(nng_msg_append_u32(m, 2u));
+	NUTS_PASS(nng_msg_append_u32(m, 3u));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000003u));
+	NUTS_PASS(nng_msg_append(m, "FAIL4", 6));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	// So on receive we should see 80000001 and 80000003.
+	NUTS_PASS(nng_recvmsg(resp, &m, 0));
+	NUTS_TRUE(nng_msg_header_len(m) == 12);
+	NUTS_TRUE(nng_msg_len(m) == 6);
+	NUTS_MATCH(nng_msg_body(m), "PASS1");
+	nng_msg_free(m);
+
+	NUTS_PASS(nng_recvmsg(resp, &m, 0));
+	NUTS_TRUE(nng_msg_header_len(m) == 16); // 3 hops + ID
+	NUTS_TRUE(nng_msg_len(m) == 6);
+	NUTS_MATCH(nng_msg_body(m), "PASS3");
+	nng_msg_free(m);
+
+	NUTS_FAIL(nng_recvmsg(resp, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+NUTS_TESTS = {
+	{ "xrespond identity", test_xresp_identity },
+	{ "xrespond raw", test_xresp_raw },
+	{ "xrespond no context", test_xresp_no_context },
+	{ "xrespond poll readable", test_xresp_poll_readable },
+	{ "xrespond poll writable", test_xresp_poll_writeable },
+	{ "xrespond validate peer", test_xresp_validate_peer },
+	{ "xrespond close pipe before send",
+	    test_xresp_close_pipe_before_send },
+	{ "xrespond close pipe during send",
+	    test_xresp_close_pipe_during_send },
+	{ "xrespond close during recv", test_xresp_close_during_recv },
+	{ "xrespond recv aio stopped", test_xresp_recv_aio_stopped },
+	{ "xrespond send no header", test_xresp_send_no_header },
+	{ "xrespond recv garbage", test_xresp_recv_garbage },
+	{ "xrespond ttl option", test_xresp_ttl_option },
+	{ "xrespond ttl drop", test_xresp_ttl_drop },
+	{ NULL, NULL },
+};
diff --git a/src/sp/protocol/survey0/xsurvey.c b/src/sp/protocol/survey0/xsurvey.c
new file mode 100644
index 00000000..2a198662
--- /dev/null
+++ b/src/sp/protocol/survey0/xsurvey.c
@@ -0,0 +1,379 @@
+//
+// 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 "core/nng_impl.h"
+#include "nng/protocol/survey0/survey.h"
+
+// Surveyor protocol.  The SURVEYOR protocol is the "survey" side of the
+// survey pattern.  This is useful for building service discovery, voting, etc.
+
+typedef struct xsurv0_pipe xsurv0_pipe;
+typedef struct xsurv0_sock xsurv0_sock;
+
+static void xsurv0_sock_getq_cb(void *);
+static void xsurv0_getq_cb(void *);
+static void xsurv0_putq_cb(void *);
+static void xsurv0_send_cb(void *);
+static void xsurv0_recv_cb(void *);
+
+// surv0_sock is our per-socket protocol private structure.
+struct xsurv0_sock {
+	nni_list       pipes;
+	nni_aio        aio_getq;
+	nni_msgq *     uwq;
+	nni_msgq *     urq;
+	nni_mtx        mtx;
+	nni_atomic_int ttl;
+};
+
+// surv0_pipe is our per-pipe protocol private structure.
+struct xsurv0_pipe {
+	nni_pipe *    npipe;
+	xsurv0_sock * psock;
+	nni_msgq *    sendq;
+	nni_list_node node;
+	nni_aio       aio_getq;
+	nni_aio       aio_putq;
+	nni_aio       aio_send;
+	nni_aio       aio_recv;
+};
+
+static void
+xsurv0_sock_fini(void *arg)
+{
+	xsurv0_sock *s = arg;
+
+	nni_aio_fini(&s->aio_getq);
+	nni_mtx_fini(&s->mtx);
+}
+
+static int
+xsurv0_sock_init(void *arg, nni_sock *nsock)
+{
+	xsurv0_sock *s = arg;
+
+	nni_aio_init(&s->aio_getq, xsurv0_sock_getq_cb, s);
+	NNI_LIST_INIT(&s->pipes, xsurv0_pipe, node);
+	nni_mtx_init(&s->mtx);
+
+	s->uwq = nni_sock_sendq(nsock);
+	s->urq = nni_sock_recvq(nsock);
+	nni_atomic_init(&s->ttl);
+	nni_atomic_set(&s->ttl, 8);
+
+	return (0);
+}
+
+static void
+xsurv0_sock_open(void *arg)
+{
+	xsurv0_sock *s = arg;
+
+	nni_msgq_aio_get(s->uwq, &s->aio_getq);
+}
+
+static void
+xsurv0_sock_close(void *arg)
+{
+	xsurv0_sock *s = arg;
+
+	nni_aio_close(&s->aio_getq);
+}
+
+static void
+xsurv0_pipe_stop(void *arg)
+{
+	xsurv0_pipe *p = arg;
+
+	nni_aio_stop(&p->aio_getq);
+	nni_aio_stop(&p->aio_send);
+	nni_aio_stop(&p->aio_recv);
+	nni_aio_stop(&p->aio_putq);
+}
+
+static void
+xsurv0_pipe_fini(void *arg)
+{
+	xsurv0_pipe *p = arg;
+
+	nni_aio_fini(&p->aio_getq);
+	nni_aio_fini(&p->aio_send);
+	nni_aio_fini(&p->aio_recv);
+	nni_aio_fini(&p->aio_putq);
+	nni_msgq_fini(p->sendq);
+}
+
+static int
+xsurv0_pipe_init(void *arg, nni_pipe *npipe, void *s)
+{
+	xsurv0_pipe *p = arg;
+	int          rv;
+
+	nni_aio_init(&p->aio_getq, xsurv0_getq_cb, p);
+	nni_aio_init(&p->aio_putq, xsurv0_putq_cb, p);
+	nni_aio_init(&p->aio_send, xsurv0_send_cb, p);
+	nni_aio_init(&p->aio_recv, xsurv0_recv_cb, p);
+
+	// This depth could be tunable.  The queue exists so that if we
+	// have multiple requests coming in faster than we can deliver them,
+	// we try to avoid dropping them.  We don't really have a solution
+	// for applying back pressure.  It would be nice if surveys carried
+	// an expiration with them, so that we could discard any that are
+	// not delivered before their expiration date.
+	if ((rv = nni_msgq_init(&p->sendq, 16)) != 0) {
+		xsurv0_pipe_fini(p);
+		return (rv);
+	}
+
+	p->npipe = npipe;
+	p->psock = s;
+	return (0);
+}
+
+static int
+xsurv0_pipe_start(void *arg)
+{
+	xsurv0_pipe *p = arg;
+	xsurv0_sock *s = p->psock;
+
+	if (nni_pipe_peer(p->npipe) != NNG_SURVEYOR0_PEER) {
+		return (NNG_EPROTO);
+	}
+
+	nni_mtx_lock(&s->mtx);
+	nni_list_append(&s->pipes, p);
+	nni_mtx_unlock(&s->mtx);
+
+	nni_msgq_aio_get(p->sendq, &p->aio_getq);
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+	return (0);
+}
+
+static void
+xsurv0_pipe_close(void *arg)
+{
+	xsurv0_pipe *p = arg;
+	xsurv0_sock *s = p->psock;
+
+	nni_aio_close(&p->aio_getq);
+	nni_aio_close(&p->aio_send);
+	nni_aio_close(&p->aio_recv);
+	nni_aio_close(&p->aio_putq);
+
+	nni_msgq_close(p->sendq);
+
+	nni_mtx_lock(&s->mtx);
+	if (nni_list_active(&s->pipes, p)) {
+		nni_list_remove(&s->pipes, p);
+	}
+	nni_mtx_unlock(&s->mtx);
+}
+
+static void
+xsurv0_getq_cb(void *arg)
+{
+	xsurv0_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_getq) != 0) {
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	nni_aio_set_msg(&p->aio_send, nni_aio_get_msg(&p->aio_getq));
+	nni_aio_set_msg(&p->aio_getq, NULL);
+
+	nni_pipe_send(p->npipe, &p->aio_send);
+}
+
+static void
+xsurv0_send_cb(void *arg)
+{
+	xsurv0_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_send) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_send));
+		nni_aio_set_msg(&p->aio_send, NULL);
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	nni_msgq_aio_get(p->sendq, &p->aio_getq);
+}
+
+static void
+xsurv0_putq_cb(void *arg)
+{
+	xsurv0_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_putq) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_putq));
+		nni_aio_set_msg(&p->aio_putq, NULL);
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	nni_pipe_recv(p->npipe, &p->aio_recv);
+}
+
+static void
+xsurv0_recv_cb(void *arg)
+{
+	xsurv0_pipe *p = arg;
+	nni_msg *    msg;
+	bool         end;
+
+	if (nni_aio_result(&p->aio_recv) != 0) {
+		nni_pipe_close(p->npipe);
+		return;
+	}
+
+	msg = nni_aio_get_msg(&p->aio_recv);
+	nni_aio_set_msg(&p->aio_recv, NULL);
+	nni_msg_set_pipe(msg, nni_pipe_id(p->npipe));
+	end = false;
+
+	while (!end) {
+		uint8_t *body;
+
+		if (nni_msg_len(msg) < 4) {
+			// Peer gave us garbage, so kick it.
+			nni_msg_free(msg);
+			nni_pipe_close(p->npipe);
+			return;
+		}
+		body = nni_msg_body(msg);
+		end  = ((body[0] & 0x80u) != 0);
+
+		if (nni_msg_header_append(msg, body, sizeof(uint32_t)) != 0) {
+			// TODO: bump a no-memory stat
+			nni_msg_free(msg);
+			// Closing the pipe may release some memory.
+			// It at least gives an indication to the peer
+			// that we've lost the message.
+			nni_pipe_close(p->npipe);
+			return;
+		}
+		nni_msg_trim(msg, sizeof(uint32_t));
+	}
+
+	nni_aio_set_msg(&p->aio_putq, msg);
+	nni_msgq_aio_put(p->psock->urq, &p->aio_putq);
+}
+
+static int
+xsurv0_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	xsurv0_sock *s = arg;
+	int          ttl;
+	int          rv;
+	if ((rv = nni_copyin_int(&ttl, buf, sz, 1, NNI_MAX_MAX_TTL, t)) == 0) {
+		nni_atomic_set(&s->ttl, ttl);
+	}
+	return (rv);
+}
+
+static int
+xsurv0_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	xsurv0_sock *s = arg;
+	return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t));
+}
+
+static void
+xsurv0_sock_getq_cb(void *arg)
+{
+	xsurv0_sock *s = arg;
+	xsurv0_pipe *p;
+	nni_msg *    msg;
+
+	if (nni_aio_result(&s->aio_getq) != 0) {
+		// Should be NNG_ECLOSED.
+		return;
+	}
+	msg = nni_aio_get_msg(&s->aio_getq);
+	nni_aio_set_msg(&s->aio_getq, NULL);
+
+	nni_mtx_lock(&s->mtx);
+	NNI_LIST_FOREACH (&s->pipes, p) {
+		nni_msg_clone(msg);
+		if (nni_msgq_tryput(p->sendq, msg) != 0) {
+			nni_msg_free(msg);
+		}
+	}
+
+	nni_msgq_aio_get(s->uwq, &s->aio_getq);
+	nni_mtx_unlock(&s->mtx);
+
+	// If there were no pipes to send on, just toss the message.
+	nni_msg_free(msg);
+}
+
+static void
+xsurv0_sock_recv(void *arg, nni_aio *aio)
+{
+	xsurv0_sock *s = arg;
+
+	nni_msgq_aio_get(s->urq, aio);
+}
+
+static void
+xsurv0_sock_send(void *arg, nni_aio *aio)
+{
+	xsurv0_sock *s = arg;
+
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static nni_proto_pipe_ops xsurv0_pipe_ops = {
+	.pipe_size  = sizeof(xsurv0_pipe),
+	.pipe_init  = xsurv0_pipe_init,
+	.pipe_fini  = xsurv0_pipe_fini,
+	.pipe_start = xsurv0_pipe_start,
+	.pipe_close = xsurv0_pipe_close,
+	.pipe_stop  = xsurv0_pipe_stop,
+};
+
+static nni_option xsurv0_sock_options[] = {
+	{
+	    .o_name = NNG_OPT_MAXTTL,
+	    .o_get  = xsurv0_sock_get_max_ttl,
+	    .o_set  = xsurv0_sock_set_max_ttl,
+	},
+	// terminate list
+	{
+	    .o_name = NULL,
+	},
+};
+
+static nni_proto_sock_ops xsurv0_sock_ops = {
+	.sock_size    = sizeof(xsurv0_sock),
+	.sock_init    = xsurv0_sock_init,
+	.sock_fini    = xsurv0_sock_fini,
+	.sock_open    = xsurv0_sock_open,
+	.sock_close   = xsurv0_sock_close,
+	.sock_send    = xsurv0_sock_send,
+	.sock_recv    = xsurv0_sock_recv,
+	.sock_options = xsurv0_sock_options,
+};
+
+static nni_proto xsurv0_proto = {
+	.proto_version  = NNI_PROTOCOL_VERSION,
+	.proto_self     = { NNG_SURVEYOR0_SELF, NNG_SURVEYOR0_SELF_NAME },
+	.proto_peer     = { NNG_SURVEYOR0_PEER, NNG_SURVEYOR0_PEER_NAME },
+	.proto_flags    = NNI_PROTO_FLAG_SNDRCV | NNI_PROTO_FLAG_RAW,
+	.proto_sock_ops = &xsurv0_sock_ops,
+	.proto_pipe_ops = &xsurv0_pipe_ops,
+};
+
+int
+nng_surveyor0_open_raw(nng_socket *sidp)
+{
+	return (nni_proto_open(sidp, &xsurv0_proto));
+}
diff --git a/src/sp/protocol/survey0/xsurvey_test.c b/src/sp/protocol/survey0/xsurvey_test.c
new file mode 100644
index 00000000..f8e9d401
--- /dev/null
+++ b/src/sp/protocol/survey0/xsurvey_test.c
@@ -0,0 +1,399 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+//
+// 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 <nuts.h>
+
+static void
+test_xsurveyor_identity(void)
+{
+	nng_socket s;
+	int        p;
+	char *     n;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&s));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p));
+	NUTS_TRUE(p == NNG_SURVEYOR0_SELF); // 0x62
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PEER, &p));
+	NUTS_TRUE(p == NNG_SURVEYOR0_PEER); // 0x62
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PROTONAME, &n));
+	NUTS_MATCH(n, NNG_SURVEYOR0_SELF_NAME);
+	nng_strfree(n);
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PEERNAME, &n));
+	NUTS_MATCH(n, NNG_SURVEYOR0_PEER_NAME);
+	nng_strfree(n);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_xsurveyor_raw(void)
+{
+	nng_socket s;
+	bool       b;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&s));
+	NUTS_PASS(nng_socket_get_bool(s, NNG_OPT_RAW, &b));
+	NUTS_TRUE(b);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_xsurvey_no_context(void)
+{
+	nng_socket s;
+	nng_ctx    ctx;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&s));
+	NUTS_FAIL(nng_ctx_open(&ctx, s), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_xsurvey_poll_writeable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_get_int(surv, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Survey is broadcast, so we can always write.
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	NUTS_MARRY(surv, resp);
+
+	// Now it's writable.
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xsurvey_poll_readable(void)
+{
+	int        fd;
+	nng_socket surv;
+	nng_socket resp;
+	nng_msg *  msg;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_respondent0_open(&resp));
+	NUTS_PASS(nng_socket_get_int(surv, NNG_OPT_RECVFD, &fd));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_TRUE(fd >= 0);
+
+	// Not readable if not connected!
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// Even after connect (no message yet)
+	NUTS_MARRY(surv, resp);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// But once we send messages, it is.
+	// We have to send a request, in order to send a reply.
+	NUTS_PASS(nng_msg_alloc(&msg, 0));
+	// Request ID
+	NUTS_PASS(nng_msg_append_u32(msg, 0x80000000));
+	NUTS_PASS(nng_sendmsg(surv, msg, 0));
+
+	NUTS_PASS(nng_recvmsg(resp, &msg, 0));
+	NUTS_PASS(nng_sendmsg(resp, msg, 0));
+
+	NUTS_SLEEP(100);
+
+	NUTS_TRUE(nuts_poll_fd(fd) );
+
+	// and receiving makes it no longer ready
+	NUTS_PASS(nng_recvmsg(surv, &msg, 0));
+	nng_msg_free(msg);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xsurvey_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char       *addr;
+
+	NUTS_ADDR(addr, "inproc");
+
+	NUTS_PASS(nng_surveyor0_open_raw(&s1));
+	NUTS_PASS(nng_surveyor0_open(&s2));
+
+	NUTS_PASS(nng_listen(s1, addr, NULL, 0));
+	NUTS_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	NUTS_SLEEP(100);
+	NUTS_PASS(nng_stats_get(&stats));
+
+	NUTS_TRUE(stats != NULL);
+	NUTS_TRUE((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	NUTS_TRUE((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	NUTS_TRUE(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	NUTS_TRUE(nng_stat_value(reject) > 0);
+
+	NUTS_CLOSE(s1);
+	NUTS_CLOSE(s2);
+	nng_stats_free(stats);
+}
+
+static void
+test_xsurvey_recv_aio_stopped(void)
+{
+	nng_socket surv;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_stop(aio);
+	nng_recv_aio(surv, aio);
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_CLOSE(surv);
+	nng_aio_free(aio);
+}
+
+static void
+test_xsurvey_recv_garbage(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+	uint32_t   req_id;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY(surv, resp);
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append_u32(m, 0x80000000));
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+
+	NUTS_PASS(nng_recvmsg(resp, &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.
+	NUTS_TRUE(nng_msg_header_len(m) == 8);
+	NUTS_PASS(nng_msg_header_chop_u32(m, &req_id));
+	NUTS_TRUE(req_id == 0x80000000);
+
+	NUTS_PASS(nng_sendmsg(resp, m, 0));
+	NUTS_FAIL(nng_recvmsg(surv, &m, 0), NNG_ETIMEDOUT);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xsurvey_recv_header(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+	nng_pipe   p;
+	uint32_t   id;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_SENDTIMEO, 1000));
+
+	NUTS_MARRY_EX(surv, resp, NULL, NULL, &p);
+
+	// Simulate a few hops.
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_header_append_u32(m, nng_pipe_id(p)));
+	NUTS_PASS(nng_msg_header_append_u32(m, 0x2));
+	NUTS_PASS(nng_msg_header_append_u32(m, 0x1));
+	NUTS_PASS(nng_msg_header_append_u32(m, 0x80000123u));
+
+	NUTS_PASS(nng_sendmsg(resp, m, 0));
+
+	NUTS_PASS(nng_recvmsg(surv, &m, 0));
+	NUTS_TRUE(nng_msg_header_len(m) == 12);
+	NUTS_PASS(nng_msg_header_trim_u32(m, &id));
+	NUTS_TRUE(id == 0x2);
+	NUTS_PASS(nng_msg_header_trim_u32(m, &id));
+	NUTS_TRUE(id == 0x1);
+	NUTS_PASS(nng_msg_header_trim_u32(m, &id));
+	NUTS_TRUE(id == 0x80000123u);
+
+	nng_msg_free(m);
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xsurvey_close_during_recv(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+	nng_pipe   p1;
+	nng_pipe   p2;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 100));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_RECVBUF, 1));
+	NUTS_PASS(nng_socket_set_int(resp, NNG_OPT_SENDBUF, 20));
+
+	NUTS_MARRY_EX(surv, resp, NULL, &p1, &p2);
+	NUTS_TRUE(nng_pipe_id(p1) > 0);
+	NUTS_TRUE(nng_pipe_id(p2) > 0);
+
+	for (unsigned i = 0; i < 20; i++) {
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_header_append_u32(m, nng_pipe_id(p2)));
+		NUTS_PASS(nng_msg_header_append_u32(m, i | 0x80000000u));
+		NUTS_SLEEP(10);
+		NUTS_PASS(nng_sendmsg(resp, m, 0));
+	}
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xsurvey_close_pipe_during_send(void)
+{
+	nng_socket resp;
+	nng_socket surv;
+	nng_msg *  m;
+	nng_pipe   p1;
+	nng_pipe   p2;
+
+	NUTS_PASS(nng_respondent0_open_raw(&resp));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(resp, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 100));
+	NUTS_PASS(nng_socket_set_int(resp, NNG_OPT_RECVBUF, 5));
+	NUTS_PASS(nng_socket_set_int(surv, NNG_OPT_SENDBUF, 20));
+
+	NUTS_MARRY_EX(surv, resp, NULL, &p1, &p2);
+	NUTS_TRUE(nng_pipe_id(p1) > 0);
+	NUTS_TRUE(nng_pipe_id(p2) > 0);
+
+	for (unsigned i = 0; i < 20; i++) {
+		NUTS_PASS(nng_msg_alloc(&m, 4));
+		NUTS_PASS(nng_msg_header_append_u32(m, i | 0x80000000u));
+		NUTS_SLEEP(10);
+		NUTS_PASS(nng_sendmsg(surv, m, 0));
+	}
+
+	NUTS_PASS(nng_pipe_close(p1));
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp);
+}
+
+static void
+test_xsurvey_ttl_option(void)
+{
+	nng_socket  s;
+	int         v;
+	bool        b;
+	size_t      sz;
+	const char *opt = NNG_OPT_MAXTTL;
+
+	NUTS_PASS(nng_surveyor0_open_raw(&s));
+
+	NUTS_PASS(nng_socket_set_int(s, opt, 1));
+	NUTS_FAIL(nng_socket_set_int(s, opt, 0), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(s, opt, -1), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(s, opt, 16), NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_int(s, opt, 256), NNG_EINVAL);
+	NUTS_PASS(nng_socket_set_int(s, opt, 3));
+	NUTS_PASS(nng_socket_get_int(s, opt, &v));
+	NUTS_TRUE(v == 3);
+	v  = 0;
+	sz = sizeof(v);
+	NUTS_PASS(nng_socket_get(s, opt, &v, &sz));
+	NUTS_TRUE(v == 3);
+	NUTS_TRUE(sz == sizeof(v));
+
+	NUTS_FAIL(nng_socket_set(s, opt, "", 1) , NNG_EINVAL);
+	sz = 1;
+	NUTS_FAIL(nng_socket_get(s, opt, &v, &sz) , NNG_EINVAL);
+	NUTS_FAIL(nng_socket_set_bool(s, opt, true) , NNG_EBADTYPE);
+	NUTS_FAIL(nng_socket_get_bool(s, opt, &b) , NNG_EBADTYPE);
+
+	NUTS_CLOSE(s);
+}
+
+static void
+test_xsurvey_broadcast(void)
+{
+	nng_socket resp1;
+	nng_socket resp2;
+	nng_socket surv;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_respondent0_open(&resp1));
+	NUTS_PASS(nng_respondent0_open(&resp2));
+	NUTS_PASS(nng_surveyor0_open_raw(&surv));
+	NUTS_PASS(nng_socket_set_ms(resp1, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(resp2, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(surv, NNG_OPT_SENDTIMEO, 100));
+
+	NUTS_MARRY(surv, resp1);
+	NUTS_MARRY(surv, resp2);
+
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_header_append_u32(m, 0x80000002u));
+	NUTS_PASS(nng_msg_append(m, "hello", 6));
+
+	NUTS_PASS(nng_sendmsg(surv, m, 0));
+	NUTS_RECV(resp1, "hello");
+	NUTS_RECV(resp2, "hello");
+
+	NUTS_CLOSE(surv);
+	NUTS_CLOSE(resp1);
+	NUTS_CLOSE(resp2);
+}
+
+TEST_LIST = {
+	{ "xsurvey identity", test_xsurveyor_identity },
+	{ "xsurvey raw", test_xsurveyor_raw },
+	{ "xsurvey no context", test_xsurvey_no_context },
+	{ "xsurvey poll readable", test_xsurvey_poll_readable },
+	{ "xsurvey poll writable", test_xsurvey_poll_writeable },
+	{ "xsurvey validate peer", test_xsurvey_validate_peer },
+	{ "xsurvey recv aio stopped", test_xsurvey_recv_aio_stopped },
+	{ "xsurvey recv garbage", test_xsurvey_recv_garbage },
+	{ "xsurvey recv header", test_xsurvey_recv_header },
+	{ "xsurvey close during recv", test_xsurvey_close_during_recv },
+	{ "xsurvey close pipe during send",
+	    test_xsurvey_close_pipe_during_send },
+	{ "xsurvey ttl option", test_xsurvey_ttl_option },
+	{ "xsurvey broadcast", test_xsurvey_broadcast },
+	{ NULL, NULL },
+};