fixes #919 Polling on subscriber socket recvfd seems broken

sub0_recv_cb was not calling nni_pollable_raise on sock->recvable.

3 files changed, 139 insertions, 2 deletions

diff --git a/src/protocol/pubsub0/sub.c b/src/protocol/pubsub0/sub.c
index fefd79a9..9c71fb04 100644
--- a/src/protocol/pubsub0/sub.c
+++ b/src/protocol/pubsub0/sub.c
@@ -10,6 +10,7 @@
 
 #include <stdlib.h>
 #include <string.h>
+#include <stdbool.h>
 
 #include "core/nng_impl.h"
 #include "nng/protocol/pubsub0/sub.h"
@@ -353,6 +354,7 @@ sub0_recv_cb(void *arg)
 	uint8_t *  body;
 	nni_list   finish;
 	nng_aio *  aio;
+	bool       submatch;
 
 	if (nni_aio_result(p->aio_recv) != 0) {
 		nni_pipe_close(p->pipe);
@@ -365,8 +367,9 @@ sub0_recv_cb(void *arg)
 	nni_aio_set_msg(p->aio_recv, NULL);
 	nni_msg_set_pipe(msg, nni_pipe_id(p->pipe));
 
-	body = nni_msg_body(msg);
-	len  = nni_msg_len(msg);
+	body     = nni_msg_body(msg);
+	len      = nni_msg_len(msg);
+	submatch = false;
 
 	nni_mtx_lock(&sock->lk);
 	// Go through all contexts.  We will try to send up.
@@ -391,6 +394,10 @@ sub0_recv_cb(void *arg)
 			continue; // TODO: Bump a stat!
 		}
 
+		// If we got to this point, we are capable of receiving this message
+		// and it is intended for us.
+		submatch = true;
+
 		if (!nni_list_empty(&ctx->raios)) {
 			nni_aio *aio = nni_list_first(&ctx->raios);
 			nni_list_remove(&ctx->raios, aio);
@@ -415,6 +422,10 @@ sub0_recv_cb(void *arg)
 		nni_msg_free(msg);
 	}
 
+	if (submatch) {
+		nni_pollable_raise(sock->recvable);
+	}
+
 	nni_pipe_recv(p->pipe, p->aio_recv);
 }
 
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 99eceaaf..60bfed2e 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -160,6 +160,7 @@ add_nng_test(options 5)
 add_nng_test(pipe 5)
 add_nng_test(platform 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)
diff --git a/tests/pubsubpollfd.c b/tests/pubsubpollfd.c
new file mode 100644
index 00000000..6a41d5f9
--- /dev/null
+++ b/tests/pubsubpollfd.c
@@ -0,0 +1,125 @@
+//
+// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2018 Capitar IT Group BV <info@capitar.com>
+// Copyright 2019 Behrooze Sirang <behrooze@gmail.com>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <string.h>
+
+#ifndef _WIN32
+#include <poll.h>
+#include <unistd.h>
+#define INVALID_SOCKET -1
+#else
+
+#define poll WSAPoll
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif
+
+#include <windows.h>
+#include <winsock2.h>
+
+#include <mswsock.h>
+#include <ws2tcpip.h>
+
+#endif
+
+#include <nng/nng.h>
+#include <nng/protocol/pubsub0/sub.h>
+#include <nng/protocol/pubsub0/pub.h>
+#include <nng/supplemental/util/platform.h>
+
+#include "convey.h"
+#include "stubs.h"
+
+TestMain("PUBSUB pollable", {
+	Convey("Given a connected pair of sockets", {
+		nng_socket s1;
+		nng_socket s2;
+
+		So(nng_pub0_open(&s1) == 0);
+		So(nng_sub0_open(&s2) == 0);
+		Reset({
+			nng_close(s1);
+			nng_close(s2);
+		});
+		So(nng_listen(s1, "inproc://yeahbaby", NULL, 0) == 0);
+		So(nng_dial(s2, "inproc://yeahbaby", NULL, 0) == 0);
+		So(nng_setopt(s2, NNG_OPT_SUB_SUBSCRIBE, "foo", 3) == 0);
+		nng_msleep(50);
+
+		Convey("We can get a recv FD", {
+			int    fd;
+			size_t sz;
+
+			sz = sizeof(fd);
+			So(nng_getopt(s2, NNG_OPT_RECVFD, &fd, &sz) == 0);
+			So(fd != (int) INVALID_SOCKET);
+
+			Convey("And it is always the same fd", {
+				int fd2;
+				sz = sizeof(fd2);
+				So(nng_getopt(s2, NNG_OPT_RECVFD, &fd2, &sz) ==
+				    0);
+				So(fd2 == fd);
+			});
+
+			Convey("And they start non pollable", {
+				struct pollfd pfd;
+				pfd.fd      = fd;
+				pfd.events  = POLLIN;
+				pfd.revents = 0;
+
+				So(poll(&pfd, 1, 0) == 0);
+				So(pfd.revents == 0);
+			});
+
+			Convey("But if we write to subscribed topic they are pollable", {
+				struct pollfd pfd;
+				pfd.fd      = fd;
+				pfd.events  = POLLIN;
+				pfd.revents = 0;
+
+				So(nng_send(s1, "foo:kick", 8, 0) == 0);
+				So(poll(&pfd, 1, 100) == 1);
+				So((pfd.revents & POLLIN) != 0);
+			});
+			Convey("And if the topic doesn't match, it is not pollable", {
+				struct pollfd pfd;
+				pfd.fd      = fd;
+				pfd.events  = POLLIN;
+				pfd.revents = 0;
+
+				So(nng_send(s1, "bar:kick", 8, 100) == 0);
+				So(poll(&pfd, 1, 0) == 0);
+			});
+		});
+
+		Convey("We can get a send FD", {
+			int    fd;
+			size_t sz;
+
+			sz = sizeof(fd);
+			So(nng_getopt(s1, NNG_OPT_SENDFD, &fd, &sz) == 0);
+			So(fd != (int) INVALID_SOCKET);
+			So(nng_send(s1, "oops", 4, 0) == 0);
+		});
+
+		Convey("Must have a big enough size", {
+			int    fd;
+			size_t sz;
+			sz = 1;
+			So(nng_getopt(s2, NNG_OPT_RECVFD, &fd, &sz) ==
+			    NNG_EINVAL);
+			sz = 128;
+			So(nng_getopt(s2, NNG_OPT_RECVFD, &fd, &sz) == 0);
+			So(sz == sizeof(fd));
+		});
+	});
+})