fixes #863 socket activation: for TCP and IPC (POSIX only)

This introduces a new option "NNG_OPT_LISTEN_FD", understood by TCP, TLS,
and (on POSIX systems) IPC.  This option is used to pass a file descriptor
or handle (Windows) that is already listening (ready for ACCEPT to be called).

For TCP and TLS, the socket must be of type AF_INET or AF_INET6, and for IPC
it must be of type AF_UNIX.

1 files changed, 251 insertions, 0 deletions

diff --git a/src/platform/tcp_stream_test.c b/src/platform/tcp_stream_test.c
index e28fb46c..a60cf1bd 100644
--- a/src/platform/tcp_stream_test.c
+++ b/src/platform/tcp_stream_test.c
@@ -172,10 +172,261 @@ test_tcp_listen_port_zero_not_bound(void)
 	nng_stream_listener_free(l);
 }
 
+void
+test_tcp_listen_empty_address(void)
+{
+	nng_stream_listener *l;
+
+	// start a listening stream listener but do not call accept
+	NUTS_PASS(nng_stream_listener_alloc(&l, "tcp4://"));
+	NUTS_PASS(nng_stream_listener_listen(l));
+	nng_stream_listener_free(l);
+}
+
+void
+test_tcp_listen_activation(void)
+{
+#if defined(NNG_PLATFORM_POSIX)
+	nng_stream_listener *l1;
+	nng_stream_listener *l2;
+	char                *addr;
+	int                  fd;
+	int                  port;
+	nng_aio             *aio1;
+	nng_aio             *aio2;
+	nng_stream_dialer   *d;
+	nng_stream          *c1, *c2;
+	char                 url[32];
+
+	NUTS_ADDR_ZERO(addr, "tcp4");
+	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
+	NUTS_PASS(nng_aio_alloc(&aio2, NULL, NULL));
+
+	nng_aio_set_timeout(aio1, 2000);
+	nng_aio_set_timeout(aio2, 2000);
+
+	NUTS_PASS(nng_stream_listener_alloc(&l1, addr));
+	NUTS_PASS(nng_stream_listener_listen(l1));
+	NUTS_PASS(
+	    nng_stream_listener_get_int(l1, NNG_OPT_TCP_BOUND_PORT, &port));
+
+	snprintf(url, sizeof(url), "tcp4://127.0.0.1:%u", port);
+	NUTS_PASS(nng_stream_dialer_alloc(&d, url));
+
+	NUTS_PASS(nng_stream_listener_get_int(l1, NNG_OPT_LISTEN_FD, &fd));
+	NUTS_PASS(nng_stream_listener_alloc(&l2, "tcp4://"));
+	NUTS_PASS(nng_stream_listener_set_int(l2, NNG_OPT_LISTEN_FD, fd));
+	nng_stream_dialer_dial(d, aio2);
+	nng_stream_listener_accept(l2, aio1);
+
+	nng_aio_wait(aio1);
+	nng_aio_wait(aio2);
+
+	NUTS_PASS(nng_aio_result(aio1));
+	NUTS_PASS(nng_aio_result(aio2));
+
+	c1 = nng_aio_get_output(aio1, 0);
+	c2 = nng_aio_get_output(aio2, 0);
+
+	char    buf1[4];
+	char    buf2[4];
+	nng_iov iov1;
+	nng_iov iov2;
+
+	iov1.iov_buf = buf1;
+	iov1.iov_len = sizeof(buf1);
+
+	iov2.iov_buf = buf2;
+	iov2.iov_len = sizeof(buf2);
+
+	nng_aio_set_iov(aio1, 1, &iov1);
+	nng_aio_set_iov(aio2, 1, &iov2);
+
+	snprintf(buf1, sizeof(buf1), "abc");
+
+	nng_stream_send(c1, aio1);
+	nng_stream_recv(c2, aio2);
+
+	nng_aio_wait(aio1);
+	nng_aio_wait(aio2);
+
+	NUTS_PASS(nng_aio_result(aio1));
+	NUTS_PASS(nng_aio_result(aio2));
+
+	NUTS_MATCH(buf1, buf2);
+
+	nng_stream_listener_free(l1);
+	nng_stream_listener_free(l2);
+	nng_stream_dialer_free(d);
+	nng_stream_free(c1);
+	nng_stream_free(c2);
+	nng_aio_free(aio1);
+	nng_aio_free(aio2);
+#elif defined(NNG_PLATFORM_WINDOWS)
+	// Windows requires that we not have created an I/O completion port yet
+	// on the incoming FD.
+	nng_stream_listener *l2;
+	SOCKET               s;
+	char                *addr;
+	int                  port;
+	nng_aio             *aio1;
+	nng_aio             *aio2;
+	nng_stream_dialer   *d;
+	nng_stream          *c1, *c2;
+	char                 url[32];
+	SOCKADDR_IN          sin;
+
+	NUTS_ADDR_ZERO(addr, "tcp4");
+	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
+	NUTS_PASS(nng_aio_alloc(&aio2, NULL, NULL));
+
+	nng_aio_set_timeout(aio1, 2000);
+	nng_aio_set_timeout(aio2, 2000);
+
+	s                   = socket(AF_INET, SOCK_STREAM, 0);
+	sin.sin_family      = AF_INET;
+	sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+	sin.sin_port        = 0;
+	int len             = sizeof(sin);
+	NUTS_ASSERT(bind(s, (SOCKADDR *) &sin, sizeof(sin)) == 0);
+	NUTS_ASSERT(getsockname(s, (SOCKADDR *) &sin, &len) == 0);
+	port = ntohs(sin.sin_port);
+	NUTS_ASSERT(listen(s, SOMAXCONN) == 0);
+
+	NUTS_PASS(nng_stream_listener_alloc(&l2, "tcp4://"));
+	NUTS_PASS(nng_stream_listener_set_int(l2, NNG_OPT_LISTEN_FD, (int) s));
+	snprintf(url, sizeof(url), "tcp://127.0.0.1:%u", port);
+	NUTS_PASS(nng_stream_dialer_alloc(&d, url));
+	nng_stream_listener_accept(l2, aio1);
+
+	nng_stream_dialer_dial(d, aio2);
+
+	nng_aio_wait(aio1);
+	NUTS_PASS(nng_aio_result(aio1));
+
+	nng_aio_wait(aio2);
+	NUTS_PASS(nng_aio_result(aio2));
+
+	c1 = nng_aio_get_output(aio1, 0);
+	c2 = nng_aio_get_output(aio2, 0);
+
+	char    buf1[4];
+	char    buf2[4];
+	nng_iov iov1;
+	nng_iov iov2;
+
+	iov1.iov_buf = buf1;
+	iov1.iov_len = sizeof(buf1);
+
+	iov2.iov_buf = buf2;
+	iov2.iov_len = sizeof(buf2);
+
+	nng_aio_set_iov(aio1, 1, &iov1);
+	nng_aio_set_iov(aio2, 1, &iov2);
+
+	snprintf(buf1, sizeof(buf1), "abc");
+
+	nng_stream_send(c1, aio1);
+	nng_stream_recv(c2, aio2);
+
+	nng_aio_wait(aio1);
+	nng_aio_wait(aio2);
+
+	NUTS_PASS(nng_aio_result(aio1));
+	NUTS_PASS(nng_aio_result(aio2));
+
+	NUTS_MATCH(buf1, buf2);
+
+	nng_stream_listener_free(l2);
+	nng_stream_dialer_free(d);
+	nng_stream_free(c1);
+	nng_stream_free(c2);
+	nng_aio_free(aio1);
+	nng_aio_free(aio2);
+#else
+	NUTS_SKIP("Not Windows or POSIX");
+#endif
+}
+
+void
+test_tcp_listen_activation_busy(void)
+{
+	nng_stream_listener *l1;
+	int                  fd;
+
+	NUTS_PASS(nng_stream_listener_alloc(&l1, "tcp://"));
+	NUTS_PASS(nng_stream_listener_listen(l1));
+	NUTS_PASS(nng_stream_listener_get_int(l1, NNG_OPT_LISTEN_FD, &fd));
+	NUTS_FAIL(
+	    nng_stream_listener_set_int(l1, NNG_OPT_LISTEN_FD, fd), NNG_EBUSY);
+	nng_stream_listener_free(l1);
+}
+
+void
+test_tcp_listen_activation_closed(void)
+{
+	nng_stream_listener *l1;
+	nng_stream_listener *l2;
+	int                  fd;
+
+	NUTS_PASS(nng_stream_listener_alloc(&l1, "tcp://"));
+	NUTS_PASS(nng_stream_listener_alloc(&l2, "tcp://"));
+	NUTS_PASS(nng_stream_listener_listen(l2));
+	NUTS_PASS(nng_stream_listener_get_int(l2, NNG_OPT_LISTEN_FD, &fd));
+	nng_stream_listener_close(l1);
+	NUTS_FAIL(nng_stream_listener_set_int(l1, NNG_OPT_LISTEN_FD, fd),
+	    NNG_ECLOSED);
+	nng_stream_listener_free(l1);
+	nng_stream_listener_free(l2);
+}
+
+void
+test_tcp_listen_activation_wrong_family(void)
+{
+
+#if !defined(NNG_PLATFORM_POSIX) || !defined(NNG_TRANSPORT_IPC)
+	NUTS_SKIP("Not posix or no IPC");
+#else
+	nng_stream_listener *l1;
+	nng_stream_listener *l2;
+	int                  fd;
+	char                *addr;
+
+	NUTS_ADDR(addr, "ipc");
+	NUTS_PASS(nng_stream_listener_alloc(&l1, "tcp://"));
+	NUTS_PASS(nng_stream_listener_alloc(&l2, addr));
+	NUTS_PASS(nng_stream_listener_listen(l2));
+	NUTS_PASS(nng_stream_listener_get_int(l2, NNG_OPT_LISTEN_FD, &fd));
+	NUTS_FAIL(nng_stream_listener_set_int(l1, NNG_OPT_LISTEN_FD, fd),
+	    NNG_EADDRINVAL);
+	nng_stream_listener_free(l1);
+	nng_stream_listener_free(l2);
+#endif
+}
+
+void
+test_tcp_listen_activation_bogus_fd(void)
+{
+	nng_stream_listener *l1;
+
+	NUTS_PASS(nng_stream_listener_alloc(&l1, "tcp://"));
+	NUTS_FAIL(nng_stream_listener_set_int(l1, NNG_OPT_LISTEN_FD, 12345),
+	    NNG_ECLOSED);
+	nng_stream_listener_free(l1);
+}
+
 NUTS_TESTS = {
 	{ "tcp stream", test_tcp_stream },
 	{ "tcp listen accept cancel", test_tcp_listen_accept_cancel },
 	{ "tcp listen port zero not bound",
 	    test_tcp_listen_port_zero_not_bound },
+	{ "tcp listen empty address", test_tcp_listen_empty_address },
+	{ "tcp socket activation", test_tcp_listen_activation },
+	{ "tcp socket activation busy", test_tcp_listen_activation_busy },
+	{ "tcp socket activation closed", test_tcp_listen_activation_closed },
+	{ "tcp socket activation wrong family",
+	    test_tcp_listen_activation_wrong_family },
+	{ "tcp socket activation bogus fd",
+	    test_tcp_listen_activation_bogus_fd },
 	{ NULL, NULL },
 };