UDP: Introduce an experimental (undocumented for now) public API for UDP. (#1838)

This exposes the UDP methods as nng_ methods, and adds support for Multicast Membership,
which is useful in a variety of situations.

No documentation is provided, and applications should consider thios API experimental.

5 files changed, 421 insertions, 98 deletions

diff --git a/src/core/platform.h b/src/core/platform.h
index 53ef53af..d7a88238 100644
--- a/src/core/platform.h
+++ b/src/core/platform.h
@@ -407,6 +407,10 @@ extern void nni_plat_udp_send(nni_plat_udp *, nni_aio *);
 // NNG_EMSGSIZE results.
 extern void nni_plat_udp_recv(nni_plat_udp *, nni_aio *);
 
+// nni_plat_udp_membership provides for joining or leaving multicast groups.
+extern int nni_plat_udp_multicast_membership(
+    nni_plat_udp *udp, nni_sockaddr *sa, bool join);
+
 //
 // Notification Pipe Pairs
 //
diff --git a/src/nng.c b/src/nng.c
index 474f07b1..e9574626 100644
--- a/src/nng.c
+++ b/src/nng.c
@@ -10,6 +10,7 @@
 
 #include "nng/nng.h"
 #include "core/nng_impl.h"
+#include "core/platform.h"
 
 // This file provides the "public" API.  This is a thin wrapper around
 // internal API functions.  We use the public prefix instead of internal,
@@ -2174,3 +2175,41 @@ nng_socket_pair(int fds[2])
 {
 	return (nni_socket_pair(fds));
 }
+
+int
+nng_udp_open(nng_udp **udp, nng_sockaddr *sa)
+{
+	(void) nni_init();
+	return (nni_plat_udp_open((nni_plat_udp **) udp, sa));
+}
+
+void
+nng_udp_close(nng_udp *udp)
+{
+	nni_plat_udp_close((nni_plat_udp *) udp);
+}
+
+int
+nng_udp_sockname(nng_udp *udp, nng_sockaddr *sa)
+{
+	return (nni_plat_udp_sockname((nni_plat_udp *) udp, sa));
+}
+
+void
+nng_udp_send(nng_udp *udp, nng_aio *aio)
+{
+	nni_plat_udp_send((nni_plat_udp *) udp, aio);
+}
+
+void
+nng_udp_recv(nng_udp *udp, nng_aio *aio)
+{
+	nni_plat_udp_recv((nni_plat_udp *) udp, aio);
+}
+
+int
+nng_udp_multicast_membership(nng_udp *udp, nng_sockaddr *sa, bool join)
+{
+	return (
+	    nni_plat_udp_multicast_membership((nni_plat_udp *) udp, sa, join));
+}
diff --git a/src/platform/posix/posix_udp.c b/src/platform/posix/posix_udp.c
index 735953fa..4ef4c68c 100644
--- a/src/platform/posix/posix_udp.c
+++ b/src/platform/posix/posix_udp.c
@@ -1,5 +1,5 @@
 //
-// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2024 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
@@ -9,6 +9,9 @@
 //
 
 #include "core/nng_impl.h"
+#include "nng/nng.h"
+#include "platform/posix/posix_impl.h"
+#include <sys/errno.h>
 
 #ifdef NNG_PLATFORM_POSIX
 #include "platform/posix/posix_pollq.h"
@@ -27,6 +30,22 @@
 #define MSG_NOSIGNAL 0
 #endif
 
+#ifndef NNG_HAVE_INET6
+#undef NNG_ENABLE_IPV6
+#endif
+
+// Linux has IPV6_ADD_MEMBERSHIP and IPV6_DROP_MEMBERSHIP
+#ifndef IPV6_JOIN_GROUP
+#ifdef IPV6_ADD_MEMBERSHIP
+#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
+#endif
+#endif
+#ifndef IPV6_LEAVE_GROUP
+#ifdef IPV6_DROP_MEMBERSHIP
+#define IPV6_LEAVE_GROUP IPV6_DROP_MEMBERSHIP
+#endif
+#endif
+
 struct nni_plat_udp {
 	nni_posix_pfd *udp_pfd;
 	int            udp_fd;
@@ -56,15 +75,15 @@ nni_posix_udp_doclose(nni_plat_udp *udp)
 static void
 nni_posix_udp_dorecv(nni_plat_udp *udp)
 {
-	nni_aio * aio;
+	nni_aio  *aio;
 	nni_list *q = &udp->udp_recvq;
 	// While we're able to recv, do so.
 	while ((aio = nni_list_first(q)) != NULL) {
 		struct iovec            iov[4];
 		unsigned                niov;
-		nni_iov *               aiov;
+		nni_iov                *aiov;
 		struct sockaddr_storage ss;
-		nng_sockaddr *          sa;
+		nng_sockaddr           *sa;
 		struct msghdr           hdr = { .msg_name = NULL };
 		int                     rv  = 0;
 		int                     cnt = 0;
@@ -102,7 +121,7 @@ nni_posix_udp_dorecv(nni_plat_udp *udp)
 static void
 nni_posix_udp_dosend(nni_plat_udp *udp)
 {
-	nni_aio * aio;
+	nni_aio  *aio;
 	nni_list *q = &udp->udp_sendq;
 
 	// While we're able to send, do so.
@@ -118,7 +137,7 @@ nni_posix_udp_dosend(nni_plat_udp *udp)
 			rv = NNG_EADDRINVAL;
 		} else {
 			unsigned     niov;
-			nni_iov *    aiov;
+			nni_iov     *aiov;
 			struct iovec iov[16];
 
 			nni_aio_get_iov(aio, &niov, &aiov);
@@ -192,7 +211,7 @@ nni_posix_udp_cb(nni_posix_pfd *pfd, unsigned events, void *arg)
 int
 nni_plat_udp_open(nni_plat_udp **upp, nni_sockaddr *bindaddr)
 {
-	nni_plat_udp *          udp;
+	nni_plat_udp           *udp;
 	int                     salen;
 	struct sockaddr_storage sa;
 	int                     rv;
@@ -323,4 +342,99 @@ nni_plat_udp_sockname(nni_plat_udp *udp, nni_sockaddr *sa)
 	return (nni_posix_sockaddr2nn(sa, &ss, sz));
 }
 
+// Joining a multicast group is different than binding to a multicast
+// group.  This allows to receive both unicast and multicast at the given
+// address.
+static int
+ip4_multicast_member(nni_plat_udp *udp, struct sockaddr *sa, bool join)
+{
+	struct ip_mreq          mreq;
+	struct sockaddr_in     *sin;
+	struct sockaddr_storage local;
+	socklen_t               sz = sizeof(local);
+
+	if (getsockname(udp->udp_fd, (struct sockaddr *) &local, &sz) >= 0) {
+		if (local.ss_family != AF_INET) {
+			// address families have to match
+			return (NNG_EADDRINVAL);
+		}
+		sin                       = (struct sockaddr_in *) &local;
+		mreq.imr_interface.s_addr = sin->sin_addr.s_addr;
+	} else {
+		mreq.imr_interface.s_addr = INADDR_ANY;
+	}
+
+	// Determine our local interface
+	sin = (struct sockaddr_in *) sa;
+
+	mreq.imr_multiaddr.s_addr = sin->sin_addr.s_addr;
+	if (setsockopt(udp->udp_fd, IPPROTO_IP,
+	        join ? IP_ADD_MEMBERSHIP : IP_DROP_MEMBERSHIP, &mreq,
+	        sizeof(mreq)) == 0) {
+		return (0);
+	}
+	return (nni_plat_errno(errno));
+}
+
+#ifdef NNG_ENABLE_IPV6
+static int
+ip6_multicast_member(nni_plat_udp *udp, struct sockaddr *sa, bool join)
+{
+	struct ipv6_mreq        mreq;
+	struct sockaddr_in6    *sin6;
+	struct sockaddr_storage local;
+	socklen_t               sz = sizeof(local);
+
+	if (getsockname(udp->udp_fd, (struct sockaddr *) &local, &sz) >= 0) {
+		if (local.ss_family != AF_INET6) {
+			// address families have to match
+			return (NNG_EADDRINVAL);
+		}
+		sin6                  = (struct sockaddr_in6 *) &local;
+		mreq.ipv6mr_interface = sin6->sin6_scope_id;
+	} else {
+		mreq.ipv6mr_interface = 0;
+	}
+
+	// Determine our local interface
+	sin6 = (struct sockaddr_in6 *) sa;
+
+	mreq.ipv6mr_multiaddr = sin6->sin6_addr;
+	if (setsockopt(udp->udp_fd, IPPROTO_IPV6,
+	        join ? IPV6_JOIN_GROUP : IPV6_LEAVE_GROUP, &mreq,
+	        sizeof(mreq)) == 0) {
+		return (0);
+	}
+	return (nni_plat_errno(errno));
+}
+#endif
+
+int
+nni_plat_udp_multicast_membership(
+    nni_plat_udp *udp, nni_sockaddr *sa, bool join)
+{
+	struct sockaddr_storage ss;
+	socklen_t               sz;
+	int                     rv;
+
+	sz = nni_posix_nn2sockaddr(&ss, sa);
+	if (sz < 1) {
+		return (NNG_EADDRINVAL);
+	}
+	switch (ss.ss_family) {
+	case AF_INET:
+		rv = ip4_multicast_member(udp, (struct sockaddr *) &ss, join);
+		break;
+#ifdef NNG_ENABLE_IPV6
+	case AF_INET6:
+		rv = ip6_multicast_member(udp, (struct sockaddr *) &ss, join);
+		break;
+#endif
+	default:
+		rv = NNG_EADDRINVAL;
+	}
+
+	return (rv);
+}
+
 #endif // NNG_PLATFORM_POSIX
diff --git a/src/platform/udp_test.c b/src/platform/udp_test.c
index 070b3eea..b6b31c9b 100644
--- a/src/platform/udp_test.c
+++ b/src/platform/udp_test.c
@@ -23,20 +23,18 @@
 void
 test_udp_pair(void)
 {
-	nng_sockaddr  sa1;
-	nng_sockaddr  sa2;
-	nni_plat_udp *u1;
-	nni_plat_udp *u2;
-	uint32_t      loopback;
-	nng_aio      *aio1;
-	nng_aio      *aio2;
-	nng_iov       iov1, iov2;
-	char          msg[] = "hello";
-	char          rbuf[1024];
-	nng_sockaddr  to;
-	nng_sockaddr  from;
-
-	NUTS_PASS(nni_init());
+	nng_sockaddr sa1;
+	nng_sockaddr sa2;
+	nng_udp     *u1;
+	nng_udp     *u2;
+	uint32_t     loopback;
+	nng_aio     *aio1;
+	nng_aio     *aio2;
+	nng_iov      iov1, iov2;
+	char         msg[] = "hello";
+	char         rbuf[1024];
+	nng_sockaddr to;
+	nng_sockaddr from;
 
 	loopback = htonl(0x7f000001); // 127.0.0.1
 
@@ -48,11 +46,11 @@ test_udp_pair(void)
 	sa2.s_in.sa_addr   = loopback;
 	sa2.s_in.sa_port   = 0;
 
-	NUTS_PASS(nni_plat_udp_open(&u1, &sa1));
-	NUTS_PASS(nni_plat_udp_open(&u2, &sa2));
+	NUTS_PASS(nng_udp_open(&u1, &sa1));
+	NUTS_PASS(nng_udp_open(&u2, &sa2));
 
-	NUTS_PASS(nni_plat_udp_sockname(u1, &sa1));
-	NUTS_PASS(nni_plat_udp_sockname(u2, &sa2));
+	NUTS_PASS(nng_udp_sockname(u1, &sa1));
+	NUTS_PASS(nng_udp_sockname(u2, &sa2));
 
 	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
 	NUTS_PASS(nng_aio_alloc(&aio2, NULL, NULL));
@@ -68,8 +66,8 @@ test_udp_pair(void)
 	NUTS_PASS(nng_aio_set_iov(aio2, 1, &iov2));
 	NUTS_PASS(nng_aio_set_input(aio2, 0, &from));
 
-	nni_plat_udp_recv(u2, aio2);
-	nni_plat_udp_send(u1, aio1);
+	nng_udp_recv(u2, aio2);
+	nng_udp_send(u1, aio1);
 	nng_aio_wait(aio1);
 	nng_aio_wait(aio2);
 
@@ -83,26 +81,24 @@ test_udp_pair(void)
 
 	nng_aio_free(aio1);
 	nng_aio_free(aio2);
-	nni_plat_udp_close(u1);
-	nni_plat_udp_close(u2);
+	nng_udp_close(u1);
+	nng_udp_close(u2);
 }
 
 void
 test_udp_multi_send_recv(void)
 {
-	nng_sockaddr  sa1, sa2, sa3, sa4;
-	nni_plat_udp *u1;
-	nni_plat_udp *u2;
-	uint32_t      loopback;
-	nng_aio      *aio1, *aio2, *aio3, *aio4;
-	nng_iov       iov1, iov2, iov3, iov4;
-	char          msg1[] = "hello";
-	char          msg2[] = "there";
-	char          rbuf1[32];
-	char          rbuf2[32];
-	nng_sockaddr  to;
-
-	NUTS_PASS(nni_init());
+	nng_sockaddr sa1, sa2, sa3, sa4;
+	nng_udp     *u1;
+	nng_udp     *u2;
+	uint32_t     loopback;
+	nng_aio     *aio1, *aio2, *aio3, *aio4;
+	nng_iov      iov1, iov2, iov3, iov4;
+	char         msg1[] = "hello";
+	char         msg2[] = "there";
+	char         rbuf1[32];
+	char         rbuf2[32];
+	nng_sockaddr to;
 
 	loopback = htonl(0x7f000001); // 127.0.0.1
 
@@ -114,11 +110,11 @@ test_udp_multi_send_recv(void)
 	sa2.s_in.sa_addr   = loopback;
 	sa2.s_in.sa_port   = 0;
 
-	NUTS_PASS(nni_plat_udp_open(&u1, &sa1));
-	NUTS_PASS(nni_plat_udp_open(&u2, &sa2));
+	NUTS_PASS(nng_udp_open(&u1, &sa1));
+	NUTS_PASS(nng_udp_open(&u2, &sa2));
 
-	NUTS_PASS(nni_plat_udp_sockname(u1, &sa1));
-	NUTS_PASS(nni_plat_udp_sockname(u2, &sa2));
+	NUTS_PASS(nng_udp_sockname(u1, &sa1));
+	NUTS_PASS(nng_udp_sockname(u2, &sa2));
 
 	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
 	NUTS_PASS(nng_aio_alloc(&aio2, NULL, NULL));
@@ -147,11 +143,11 @@ test_udp_multi_send_recv(void)
 	NUTS_PASS(nng_aio_set_iov(aio4, 1, &iov4));
 	NUTS_PASS(nng_aio_set_input(aio4, 0, &sa4));
 
-	nni_plat_udp_recv(u2, aio4);
-	nni_plat_udp_recv(u2, aio3);
-	nni_plat_udp_send(u1, aio2);
+	nng_udp_recv(u2, aio4);
+	nng_udp_recv(u2, aio3);
+	nng_udp_send(u1, aio2);
 	nng_msleep(100); // to keep order clear
-	nni_plat_udp_send(u1, aio1);
+	nng_udp_send(u1, aio1);
 	nng_aio_wait(aio1);
 	nng_aio_wait(aio2);
 	nng_aio_wait(aio3);
@@ -166,7 +162,7 @@ test_udp_multi_send_recv(void)
 	NUTS_ASSERT(strcmp(rbuf1, msg1) == 0);
 	NUTS_ASSERT(strcmp(rbuf2, msg2) == 0);
 
-	NUTS_PASS(nni_plat_udp_sockname(u1, &sa2));
+	NUTS_PASS(nng_udp_sockname(u1, &sa2));
 	NUTS_ASSERT(sa2.s_in.sa_family == sa3.s_in.sa_family);
 	NUTS_ASSERT(sa2.s_in.sa_addr == sa3.s_in.sa_addr);
 	NUTS_ASSERT(sa2.s_in.sa_port == sa3.s_in.sa_port);
@@ -179,21 +175,19 @@ test_udp_multi_send_recv(void)
 	nng_aio_free(aio2);
 	nng_aio_free(aio3);
 	nng_aio_free(aio4);
-	nni_plat_udp_close(u1);
-	nni_plat_udp_close(u2);
+	nng_udp_close(u1);
+	nng_udp_close(u2);
 }
 
 void
 test_udp_send_no_addr(void)
 {
-	nng_sockaddr  sa1;
-	nni_plat_udp *u1;
-	uint32_t      loopback;
-	nng_aio      *aio1;
-	nng_iov       iov1;
-	char          msg[] = "hello";
-
-	NUTS_PASS(nni_init());
+	nng_sockaddr sa1;
+	nng_udp     *u1;
+	uint32_t     loopback;
+	nng_aio     *aio1;
+	nng_iov      iov1;
+	char         msg[] = "hello";
 
 	loopback = htonl(0x7f000001); // 127.0.0.1
 
@@ -201,8 +195,8 @@ test_udp_send_no_addr(void)
 	sa1.s_in.sa_addr   = loopback;
 	sa1.s_in.sa_port   = 0; // wild card port binding
 
-	NUTS_PASS(nni_plat_udp_open(&u1, &sa1));
-	NUTS_PASS(nni_plat_udp_sockname(u1, &sa1));
+	NUTS_PASS(nng_udp_open(&u1, &sa1));
+	NUTS_PASS(nng_udp_sockname(u1, &sa1));
 
 	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
 
@@ -210,28 +204,26 @@ test_udp_send_no_addr(void)
 	iov1.iov_len = strlen(msg) + 1;
 	NUTS_PASS(nng_aio_set_iov(aio1, 1, &iov1));
 
-	nni_plat_udp_send(u1, aio1);
+	nng_udp_send(u1, aio1);
 	nng_aio_wait(aio1);
 
 	NUTS_FAIL(nng_aio_result(aio1), NNG_EADDRINVAL);
 
 	nng_aio_free(aio1);
-	nni_plat_udp_close(u1);
+	nng_udp_close(u1);
 }
 
 void
 test_udp_send_ipc(void)
 {
-	nng_sockaddr  sa1;
-	nng_sockaddr  sa2;
-	nni_plat_udp *u1;
-	uint32_t      loopback;
-	nng_aio      *aio1;
-	nng_iov       iov1;
-	char          msg[] = "hello";
-	int rv;
-
-	NUTS_PASS(nni_init());
+	nng_sockaddr sa1 = { 0 };
+	nng_sockaddr sa2 = { 0 };
+	nng_udp     *u1;
+	uint32_t     loopback;
+	nng_aio     *aio1;
+	nng_iov      iov1;
+	char         msg[] = "hello";
+	int          rv;
 
 	loopback = htonl(0x7f000001); // 127.0.0.1
 
@@ -242,8 +234,8 @@ test_udp_send_ipc(void)
 	sa2.s_ipc.sa_family = NNG_AF_IPC;
 	strcat(sa2.s_ipc.sa_path, "/tmp/bogus");
 
-	NUTS_PASS(nni_plat_udp_open(&u1, &sa1));
-	NUTS_PASS(nni_plat_udp_sockname(u1, &sa1));
+	NUTS_PASS(nng_udp_open(&u1, &sa1));
+	NUTS_PASS(nng_udp_sockname(u1, &sa1));
 
 	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
 
@@ -252,49 +244,126 @@ test_udp_send_ipc(void)
 	NUTS_PASS(nng_aio_set_iov(aio1, 1, &iov1));
 	NUTS_PASS(nng_aio_set_input(aio1, 0, &sa2));
 
-	nni_plat_udp_send(u1, aio1);
+	nng_udp_send(u1, aio1);
 	nng_aio_wait(aio1);
 
 	rv = nng_aio_result(aio1);
 	NUTS_ASSERT(rv == NNG_EADDRINVAL || rv == NNG_ENOTSUP);
 
 	nng_aio_free(aio1);
-	nni_plat_udp_close(u1);
+	nng_udp_close(u1);
 }
 
 void
 test_udp_bogus_bind(void)
 {
-	nni_plat_udp *u;
-	nng_sockaddr  sa;
-	int           rv;
+	nng_udp     *u;
+	nng_sockaddr sa = { 0 };
+	int          rv;
 
 	sa.s_ipc.sa_family = NNG_AF_IPC;
 	strcpy(sa.s_ipc.sa_path, "/tmp/t");
-	rv = nni_plat_udp_open(&u, &sa);
+	rv = nng_udp_open(&u, &sa);
 	// Some platforms reject IPC addresses altogether (Windows),
 	// whereas others just say it's not supported with UDP.
 	NUTS_ASSERT((rv == NNG_ENOTSUP) || (rv == NNG_EADDRINVAL));
 
 	// NULL address also bad.
-	NUTS_FAIL(nni_plat_udp_open(&u, NULL), NNG_EADDRINVAL);
+	NUTS_FAIL(nng_udp_open(&u, NULL), NNG_EADDRINVAL);
 }
 
 void
 test_udp_duplicate_bind(void)
 {
-	nni_plat_udp *u1;
-	nni_plat_udp *u2;
-	nng_sockaddr  sa;
+	nng_udp     *u1;
+	nng_udp     *u2;
+	nng_sockaddr sa = { 0 };
+
+	sa.s_in.sa_family = NNG_AF_INET;
+	sa.s_in.sa_addr   = htonl(0x7f000001);
+
+	NUTS_PASS(nng_udp_open(&u1, &sa));
+	NUTS_PASS(nng_udp_sockname(u1, &sa));
+	NUTS_FAIL(nng_udp_open(&u2, &sa), NNG_EADDRINUSE);
+	nng_udp_close(u1);
+}
+
+void
+test_udp_multicast_membership(void)
+{
+	nng_udp     *u1;
+	nng_sockaddr sa = { 0 };
+	nng_sockaddr mc = { 0 };
+
+	mc.s_in.sa_family = NNG_AF_INET;
+	mc.s_in.sa_addr   = htonl(0xe0000001); // 224.0.0.1 ... all hosts
 
 	sa.s_in.sa_family = NNG_AF_INET;
 	sa.s_in.sa_addr   = htonl(0x7f000001);
 
-	NUTS_PASS(nni_init());
-	NUTS_PASS(nni_plat_udp_open(&u1, &sa));
-	NUTS_PASS(nni_plat_udp_sockname(u1, &sa));
-	NUTS_FAIL(nni_plat_udp_open(&u2, &sa), NNG_EADDRINUSE);
-	nni_plat_udp_close(u1);
+	NUTS_PASS(nng_udp_open(&u1, &sa));
+	NUTS_PASS(nng_udp_sockname(u1, &sa));
+	NUTS_PASS(nng_udp_multicast_membership(u1, &mc, true));
+	NUTS_PASS(nng_udp_multicast_membership(u1, &mc, false));
+	nng_udp_close(u1);
+}
+
+void
+test_udp_multicast_send_recv(void)
+{
+	nng_udp     *u1;
+	nng_udp     *u2;
+	nng_sockaddr sa1 = { 0 };
+	nng_sockaddr sa2 = { 0 };
+	nng_sockaddr ra2 = { 0 };
+	nng_sockaddr mc  = { 0 };
+	char        *msg = "multi";
+	nng_iov      iov1;
+	nng_iov      iov2;
+	nng_aio     *aio1;
+	nng_aio     *aio2;
+	char         rbuf[32];
+
+	mc.s_in.sa_family = NNG_AF_INET;
+	mc.s_in.sa_addr   = htonl(0xe0000001); // 224.0.0.1 ... all hosts
+
+	sa1.s_in.sa_family = NNG_AF_INET;
+	sa1.s_in.sa_addr   = htonl(0x7f000001);
+
+	sa2.s_in.sa_family = NNG_AF_INET;
+	sa2.s_in.sa_addr   = htonl(0x7f000001);
+
+	NUTS_PASS(nng_udp_open(&u1, &sa1));
+	NUTS_PASS(nng_udp_sockname(u1, &sa1));
+	NUTS_PASS(nng_udp_open(&u2, &sa2));
+	NUTS_PASS(nng_udp_sockname(u2, &sa2));
+	NUTS_PASS(nng_udp_multicast_membership(u1, &mc, true));
+	NUTS_PASS(nng_udp_multicast_membership(u2, &mc, true));
+	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
+	NUTS_PASS(nng_aio_alloc(&aio2, NULL, NULL));
+
+	iov1.iov_buf = msg;
+	iov1.iov_len = strlen(msg) + 1;
+	NUTS_PASS(nng_aio_set_iov(aio1, 1, &iov1));
+	NUTS_PASS(nng_aio_set_input(aio1, 0, &sa2));
+
+	iov2.iov_buf = rbuf;
+	iov2.iov_len = sizeof(rbuf);
+	NUTS_PASS(nng_aio_set_iov(aio2, 1, &iov2));
+	NUTS_PASS(nng_aio_set_input(aio2, 0, &ra2));
+
+	nng_udp_recv(u2, aio2);
+	nng_udp_send(u1, aio1);
+	nng_aio_wait(aio1);
+	nng_aio_wait(aio2);
+	NUTS_PASS(nng_aio_result(aio1));
+	NUTS_PASS(nng_aio_result(aio2));
+
+	NUTS_MATCH(msg, rbuf);
+	nng_udp_close(u1);
+	nng_udp_close(u2);
+	nng_aio_free(aio1);
+	nng_aio_free(aio2);
 }
 
 #ifdef NNG_ENABLE_IPV6
@@ -302,20 +371,20 @@ void
 test_udp_send_v6_from_v4(void)
 {
 	int           rv;
-	nni_plat_udp *u1;
-	nng_sockaddr  sa;
+	nng_udp      *u1;
+	nng_sockaddr  sa = { 0 };
 	nng_aio      *aio1;
 	nng_iov       iov1;
 	char         *msg        = "nope";
 	const uint8_t google[16] = { 0x26, 0x07, 0xf8, 0xb0, 0x40, 0x07, 0x40,
 		0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x20, 0x0e };
 
+	memset(&sa, 0, sizeof(sa));
 	sa.s_in.sa_family = NNG_AF_INET;
 	sa.s_in.sa_addr   = htonl(0x7f000001);
 
-	NUTS_PASS(nni_init());
 	NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
-	NUTS_PASS(nni_plat_udp_open(&u1, &sa));
+	NUTS_PASS(nng_udp_open(&u1, &sa));
 
 	sa.s_in6.sa_family = NNG_AF_INET6;
 	memcpy(sa.s_in6.sa_addr, google, 16);
@@ -326,14 +395,14 @@ test_udp_send_v6_from_v4(void)
 	nng_aio_set_iov(aio1, 1, &iov1);
 	nng_aio_set_input(aio1, 0, &sa);
 
-	nni_plat_udp_send(u1, aio1);
+	nng_udp_send(u1, aio1);
 	nng_aio_wait(aio1);
 	rv = nng_aio_result(aio1);
 	NUTS_ASSERT((rv == NNG_EADDRINVAL) || (rv == NNG_ENOTSUP) ||
 	    (rv == NNG_EUNREACHABLE));
 
 	nng_aio_free(aio1);
-	nni_plat_udp_close(u1);
+	nng_udp_close(u1);
 }
 #endif // NNG_ENABLE_IPV6
 
@@ -344,6 +413,8 @@ NUTS_TESTS = {
 	{ "udp send ipc address", test_udp_send_ipc },
 	{ "udp bogus bind", test_udp_bogus_bind },
 	{ "udp duplicate bind", test_udp_duplicate_bind },
+	{ "udp multicast membership", test_udp_multicast_membership },
+	{ "udp multicast send recv", test_udp_multicast_send_recv },
 #ifdef NNG_ENABLE_IPV6
 	{ "udp send v6 from v4", test_udp_send_v6_from_v4 },
 #endif
diff --git a/src/platform/windows/win_udp.c b/src/platform/windows/win_udp.c
index 09064e2f..c1c7ef21 100644
--- a/src/platform/windows/win_udp.c
+++ b/src/platform/windows/win_udp.c
@@ -315,6 +315,101 @@ nni_plat_udp_sockname(nni_plat_udp *udp, nni_sockaddr *sa)
 	return (nni_win_sockaddr2nn(sa, &ss));
 }
 
+// Joining a multicast group is different than binding to a multicast
+// group.  This allows to receive both unicast and multicast at the given
+// address.
+static int
+ip4_multicast_member(nni_plat_udp *udp, SOCKADDR *sa, bool join)
+{
+	IP_MREQ          mreq;
+	SOCKADDR_IN     *sin;
+	SOCKADDR_STORAGE local;
+	int              sz = sizeof(local);
+
+	if (getsockname(udp->s, (SOCKADDR *) &local, &sz) >= 0) {
+		if (local.ss_family != AF_INET) {
+			// address families have to match
+			return (NNG_EADDRINVAL);
+		}
+		sin                       = (SOCKADDR_IN *) &local;
+		mreq.imr_interface.s_addr = sin->sin_addr.s_addr;
+	} else {
+		mreq.imr_interface.s_addr = INADDR_ANY;
+	}
+
+	// Determine our local interface
+	sin = (SOCKADDR_IN *) sa;
+
+	mreq.imr_multiaddr.s_addr = sin->sin_addr.s_addr;
+	if (setsockopt(udp->s, IPPROTO_IP,
+	        join ? IP_ADD_MEMBERSHIP : IP_DROP_MEMBERSHIP,
+	        (const char *) &mreq, sizeof(mreq)) == 0) {
+		return (0);
+	}
+	return (nni_win_error(GetLastError()));
+}
+
+#ifdef NNG_ENABLE_IPV6
+static int
+ip6_multicast_member(nni_plat_udp *udp, SOCKADDR *sa, bool join)
+{
+	IPV6_MREQ        mreq;
+	SOCKADDR_IN6    *sin6;
+	SOCKADDR_STORAGE local;
+	int              sz = sizeof(local);
+
+	if (getsockname(udp->s, (SOCKADDR *) &local, &sz) >= 0) {
+		if (local.ss_family != AF_INET6) {
+			// address families have to match
+			return (NNG_EADDRINVAL);
+		}
+		sin6                  = (SOCKADDR_IN6 *) &local;
+		mreq.ipv6mr_interface = sin6->sin6_scope_id;
+	} else {
+		mreq.ipv6mr_interface = 0;
+	}
+
+	// Determine our local interface
+	sin6 = (SOCKADDR_IN6 *) sa;
+
+	mreq.ipv6mr_multiaddr = sin6->sin6_addr;
+	if (setsockopt(udp->s, IPPROTO_IPV6,
+	        join ? IPV6_JOIN_GROUP : IPV6_LEAVE_GROUP,
+	        (const char *) &mreq, sizeof(mreq)) == 0) {
+		return (0);
+	}
+	return (nni_win_error(GetLastError()));
+}
+#endif
+
+int
+nni_plat_udp_multicast_membership(
+    nni_plat_udp *udp, nni_sockaddr *sa, bool join)
+{
+	SOCKADDR_STORAGE ss;
+	socklen_t        sz;
+	int              rv;
+
+	sz = nni_win_nn2sockaddr(&ss, sa);
+	if (sz < 1) {
+		return (NNG_EADDRINVAL);
+	}
+	switch (ss.ss_family) {
+	case AF_INET:
+		rv = ip4_multicast_member(udp, (struct sockaddr *) &ss, join);
+		break;
+#ifdef NNG_ENABLE_IPV6
+	case AF_INET6:
+		rv = ip6_multicast_member(udp, (struct sockaddr *) &ss, join);
+		break;
+#endif
+	default:
+		rv = NNG_EADDRINVAL;
+	}
+
+	return (rv);
+}
+
 int
 nni_win_udp_sysinit(void)
 {