fixes #1121 reconnect test failures (Darwin?)

2 files changed, 116 insertions, 92 deletions

diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index fd9d1f68..80ee0a45 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -127,6 +127,7 @@ endif ()
 nng_test(aio)
 nng_test(bufsz)
 nng_test(platform)
+nng_test(reconnect)
 nng_test(sock)
 
 add_nng_test(device 5)
@@ -148,7 +149,6 @@ add_nng_test(options 5)
 add_nng_test(pipe 5)
 add_nng_test(pollfd 5)
 add_nng_test(pubsubpollfd 5)
-add_nng_test(reconnect 5)
 add_nng_test1(resolv 10 NNG_STATIC_LIB)
 add_nng_test(scalability 20 ON)
 add_nng_test(set_recvmaxsize 2)
diff --git a/tests/reconnect.c b/tests/reconnect.c
index 0cc2878b..0ee63c66 100644
--- a/tests/reconnect.c
+++ b/tests/reconnect.c
@@ -1,5 +1,5 @@
 //
-// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// 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
@@ -13,93 +13,117 @@
 #include <nng/nng.h>
 #include <nng/protocol/pipeline0/pull.h>
 #include <nng/protocol/pipeline0/push.h>
-#include <nng/supplemental/util/platform.h>
-
-#include "convey.h"
-#include "stubs.h"
-
-#define APPENDSTR(m, s) nng_msg_append(m, s, strlen(s))
-#define CHECKSTR(m, s)                   \
-	So(nng_msg_len(m) == strlen(s)); \
-	So(memcmp(nng_msg_body(m), s, strlen(s)) == 0)
-
-TestMain("Reconnect works", {
-	atexit(nng_fini);
-	const char *addr = "inproc://test";
-	Convey("We can create a pipeline", {
-		nng_socket push;
-		nng_socket pull;
-
-		So(nng_push_open(&push) == 0);
-		So(nng_pull_open(&pull) == 0);
-
-		Reset({
-			nng_close(push);
-			nng_close(pull);
-		});
-
-		So(nng_setopt_ms(pull, NNG_OPT_RECONNMINT, 10) == 0);
-		So(nng_setopt_ms(pull, NNG_OPT_RECONNMAXT, 10) == 0);
-
-		Convey("Dialing before listening works", {
-			So(nng_dial(push, addr, NULL, NNG_FLAG_NONBLOCK) == 0);
-			nng_msleep(100);
-			So(nng_listen(pull, addr, NULL, 0) == 0);
-
-			Convey("We can send a frame", {
-				nng_msg *msg;
-
-				nng_msleep(100);
-				So(nng_msg_alloc(&msg, 0) == 0);
-				APPENDSTR(msg, "hello");
-				So(nng_sendmsg(push, msg, 0) == 0);
-				msg = NULL;
-				So(nng_recvmsg(pull, &msg, 0) == 0);
-				So(msg != NULL);
-				CHECKSTR(msg, "hello");
-				nng_msg_free(msg);
-			});
-		});
-		Convey("Reconnection works", {
-			nng_listener l;
-			So(nng_dial(push, addr, NULL, NNG_FLAG_NONBLOCK) == 0);
-			So(nng_listen(pull, addr, &l, 0) == 0);
-			nng_msleep(100);
-			nng_listener_close(l);
-			So(nng_listen(pull, addr, NULL, 0) == 0);
-
-			Convey("They still exchange frames", {
-				nng_msg *msg;
-				nng_pipe p1;
-
-				nng_msleep(100);
-				So(nng_msg_alloc(&msg, 0) == 0);
-				APPENDSTR(msg, "hello");
-				So(nng_sendmsg(push, msg, 0) == 0);
-				msg = NULL;
-				So(nng_recvmsg(pull, &msg, 0) == 0);
-				So(msg != NULL);
-				CHECKSTR(msg, "hello");
-				p1 = nng_msg_get_pipe(msg);
-				nng_msg_free(msg);
-
-				Convey("Even after pipe close", {
-					nng_pipe p2;
-
-					nng_pipe_close(p1);
-					nng_msleep(100);
-					So(nng_msg_alloc(&msg, 0) == 0);
-					APPENDSTR(msg, "again");
-					So(nng_sendmsg(push, msg, 0) == 0);
-					msg = NULL;
-					So(nng_recvmsg(pull, &msg, 0) == 0);
-					So(msg != NULL);
-					CHECKSTR(msg, "again");
-					p2 = nng_msg_get_pipe(msg);
-					nng_msg_free(msg);
-					So(nng_pipe_id(p2) != nng_pipe_id(p1));
-				});
-			});
-		});
-	});
-})
+
+#include "acutest.h"
+#include "testutil.h"
+
+void
+test_dial_before_listen(void)
+{
+	nng_socket push;
+	nng_socket pull;
+	char       addr[64];
+
+	testutil_scratch_addr("inproc", sizeof(addr), addr);
+
+	TEST_NNG_PASS(nng_push0_open(&push));
+	TEST_NNG_PASS(nng_pull0_open(&pull));
+
+	TEST_NNG_PASS(nng_setopt_ms(pull, NNG_OPT_RECONNMINT, 10));
+	TEST_NNG_PASS(nng_setopt_ms(pull, NNG_OPT_RECONNMAXT, 10));
+
+	TEST_NNG_PASS(nng_dial(pull, addr, NULL, NNG_FLAG_NONBLOCK));
+	testutil_sleep(100);
+	TEST_NNG_PASS(nng_listen(push, addr, NULL, 0));
+
+	TEST_NNG_SEND_STR(push, "hello");
+	TEST_NNG_RECV_STR(pull, "hello");
+
+	TEST_NNG_PASS(nng_close(push));
+	TEST_NNG_PASS(nng_close(pull));
+}
+
+void
+test_reconnect(void)
+{
+	nng_socket   push;
+	nng_socket   pull;
+	nng_listener l;
+	char         addr[64];
+
+	testutil_scratch_addr("inproc", sizeof(addr), addr);
+
+	TEST_NNG_PASS(nng_push0_open(&push));
+	TEST_NNG_PASS(nng_pull0_open(&pull));
+
+	TEST_NNG_PASS(nng_setopt_ms(pull, NNG_OPT_RECONNMINT, 10));
+	TEST_NNG_PASS(nng_setopt_ms(pull, NNG_OPT_RECONNMAXT, 10));
+
+	TEST_NNG_PASS(nng_dial(pull, addr, NULL, NNG_FLAG_NONBLOCK));
+	testutil_sleep(100);
+	TEST_NNG_PASS(nng_listen(push, addr, &l, 0));
+
+	TEST_NNG_SEND_STR(push, "hello");
+	TEST_NNG_RECV_STR(pull, "hello");
+
+	// Close the listener
+	TEST_NNG_PASS(nng_listener_close(l));
+
+	TEST_NNG_PASS(nng_listen(push, addr, &l, 0));
+	TEST_NNG_SEND_STR(push, "again");
+	TEST_NNG_RECV_STR(pull, "again");
+
+	TEST_NNG_PASS(nng_close(push));
+	TEST_NNG_PASS(nng_close(pull));
+}
+
+void
+test_reconnect_pipe(void)
+{
+	nng_socket   push;
+	nng_socket   pull;
+	nng_listener l;
+	nng_msg *msg;
+	char         addr[64];
+
+	testutil_scratch_addr("inproc", sizeof(addr), addr);
+
+	TEST_NNG_PASS(nng_push0_open(&push));
+	TEST_NNG_PASS(nng_pull0_open(&pull));
+
+	TEST_NNG_PASS(nng_setopt_ms(pull, NNG_OPT_RECONNMINT, 10));
+	TEST_NNG_PASS(nng_setopt_ms(pull, NNG_OPT_RECONNMAXT, 10));
+
+	TEST_NNG_PASS(nng_dial(pull, addr, NULL, NNG_FLAG_NONBLOCK));
+	testutil_sleep(100);
+	TEST_NNG_PASS(nng_listen(push, addr, &l, 0));
+
+	TEST_NNG_SEND_STR(push, "hello");
+
+	TEST_NNG_PASS(nng_recvmsg(pull, &msg, 0));
+	TEST_CHECK(msg != NULL);
+	TEST_CHECK(nng_msg_len(msg) == 6);
+	TEST_CHECK(strcmp(nng_msg_body(msg), "hello") == 0);
+	nng_pipe_close(nng_msg_get_pipe(msg));
+	nng_msg_free(msg);
+
+	// We have to wait a bit, because while we closed the pipe on the
+	// receiver, the receiver might not have got the update.  If we
+	// send too soon, then the message gets routed to the sender pipe
+	// that is about to close.
+	testutil_sleep(100);
+
+	// Reconnect should happen more ore less immediately.
+	TEST_NNG_SEND_STR(push, "again");
+	TEST_NNG_RECV_STR(pull, "again");
+
+	TEST_NNG_PASS(nng_close(push));
+	TEST_NNG_PASS(nng_close(pull));
+}
+
+TEST_LIST = {
+	{ "dial before listen", test_dial_before_listen },
+	{ "reconnect", test_reconnect },
+	{ "reconnect pipe", test_reconnect_pipe },
+	{ NULL, NULL },
+};
\ No newline at end of file