Repository: eclipse-vertx/vert.x

Problem statement:
`listenMulticastGroup` doesn't work with a null NetworkInterface
### Version

At least vertx 3.8.3-3.9.1

### Context

`listenMulticastGroup` throws a NullPointerException when it is called with the `listenMulticastGroup(string multicastAddress, Handler<AsyncResult<DatagramSocket>> handler)` signature.

### Do you have a reproducer?

https://github.com/hiddenswitch/udp-broadcast-reproducer

You can use `./gradlew test --tests "udp.broadcast.reproducer.AppTest.testUdpListensNullInterface"`

### Steps to reproduce

 1. Create a udp socket (socket "receipient").
 2. Listen to a multicast group with the specified overload.
 3. Observe a null pointer exception (see below).
 4. Observe that heuristically choosing an interface works.

```
SEVERE: Unhandled exception
java.lang.NullPointerException: networkInterface
	at io.netty.util.internal.ObjectUtil.checkNotNull(ObjectUtil.java:33)
	at io.netty.channel.socket.nio.NioDatagramChannel.joinGroup(NioDatagramChannel.java:409)
	at io.netty.channel.socket.nio.NioDatagramChannel.joinGroup(NioDatagramChannel.java:371)
	at io.netty.channel.socket.nio.NioDatagramChannel.joinGroup(NioDatagramChannel.java:365)
	at io.vertx.core.datagram.impl.DatagramSocketImpl.listenMulticastGroup(DatagramSocketImpl.java:96)
	at udp.broadcast.reproducer.AppTest.lambda$testUdpListensNullInterface$0(AppTest.java:31)
	at io.vertx.junit5.VertxTestContext.lambda$succeeding$1(VertxTestContext.java:182)
	at io.vertx.core.datagram.impl.DatagramSocketImpl.lambda$null$0(DatagramSocketImpl.java:206)
	at io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366)
	at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:43)
	at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:229)
	at io.vertx.core.net.impl.ChannelFutureListenerAdapter.operationComplete(ChannelFutureListenerAdapter.java:39)
	at io.vertx.core.net.impl.ChannelFutureListenerAdapter.operationComplete(ChannelFutureListenerAdapter.java:24)
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577)
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:551)
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:490)
	at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:615)
	at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:604)
	at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)
	at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:84)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.safeSetSuccess(AbstractChannel.java:984)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.bind(AbstractChannel.java:566)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.bind(DefaultChannelPipeline.java:1334)
	at io.netty.channel.AbstractChannelHandlerContext.invokeBind(AbstractChannelHandlerContext.java:506)
	at io.netty.channel.AbstractChannelHandlerContext.access$900(AbstractChannelHandlerContext.java:61)
	at io.netty.channel.AbstractChannelHandlerContext$8.run(AbstractChannelHandlerContext.java:496)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:830)
```

### Extra

**Platform:** macOS 10.15.5 (`Darwin MacBook-Pro.local 19.5.0 Darwin Kernel Version 19.5.0: Thu Apr 30 18:25:59 PDT 2020; root:xnu-6153.121.1~7/RELEASE_X86_64 x86_64`)

**Java:**
```
openjdk version "13.0.2" 2020-01-14
OpenJDK Runtime Environment (build 13.0.2+8)
OpenJDK 64-Bit Server VM (build 13.0.2+8, mixed mode, sharing)
```


Hints:
see also https://github.com/eclipse-vertx/vert.x/issues/3540
yes that one fails, the socket is not able to get a network interface from the local address

Patch:
diff --git a/src/main/java/io/vertx/core/datagram/impl/DatagramSocketImpl.java b/src/main/java/io/vertx/core/datagram/impl/DatagramSocketImpl.java
--- a/src/main/java/io/vertx/core/datagram/impl/DatagramSocketImpl.java
+++ b/src/main/java/io/vertx/core/datagram/impl/DatagramSocketImpl.java
@@ -25,6 +25,7 @@
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
+import io.vertx.core.VertxException;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.datagram.DatagramSocket;
 import io.vertx.core.datagram.DatagramSocketOptions;
@@ -43,6 +44,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.NetworkInterface;
+import java.net.SocketException;
 import java.net.UnknownHostException;
 import java.util.Objects;
 
@@ -98,11 +100,32 @@ public DatagramSocket listenMulticastGroup(String multicastAddress, Handler<Asyn
     return this;
   }
 
+  private NetworkInterface determineMulticastNetworkIface() throws Exception {
+    NetworkInterface iface = null;
+    InetSocketAddress localAddr = channel.localAddress();
+    if (localAddr != null) {
+      iface = NetworkInterface.getByInetAddress(localAddr.getAddress());
+    }
+    if (iface == null) {
+      iface = channel.config().getNetworkInterface();
+    }
+    return iface;
+  }
+
   @Override
   public Future<Void> listenMulticastGroup(String multicastAddress) {
+    NetworkInterface iface;
+    try {
+      iface = determineMulticastNetworkIface();
+    } catch (Exception e) {
+      return context.failedFuture(e);
+    }
+    if (iface == null) {
+      return context.failedFuture("A valid network interface could not be determined from the socket bind address or multicast interface");
+    }
     ChannelFuture fut;
     try {
-      fut = channel.joinGroup(InetAddress.getByName(multicastAddress));
+      fut = channel.joinGroup(InetAddress.getByName(multicastAddress), iface, null);
     } catch (UnknownHostException e) {
       return context.failedFuture(e);
     }
@@ -150,9 +173,18 @@ public DatagramSocket unlistenMulticastGroup(String multicastAddress, Handler<As
 
   @Override
   public Future<Void> unlistenMulticastGroup(String multicastAddress) {
+    NetworkInterface iface;
+    try {
+      iface = determineMulticastNetworkIface();
+    } catch (Exception e) {
+      return context.failedFuture(e);
+    }
+    if (iface == null) {
+      return context.failedFuture("A valid network interface could not be determined from the socket bind address or multicast interface");
+    }
     ChannelFuture fut;
     try {
-      fut = channel.leaveGroup(InetAddress.getByName(multicastAddress));
+      fut = channel.leaveGroup(InetAddress.getByName(multicastAddress), iface, null);
     } catch (Exception e) {
       return context.failedFuture(e);
     }


Test patch:
diff --git a/src/test/java/io/vertx/core/datagram/DatagramTest.java b/src/test/java/io/vertx/core/datagram/DatagramTest.java
--- a/src/test/java/io/vertx/core/datagram/DatagramTest.java
+++ b/src/test/java/io/vertx/core/datagram/DatagramTest.java
@@ -13,19 +13,21 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.UnpooledHeapByteBuf;
 import io.vertx.core.AbstractVerticle;
+import io.vertx.core.AsyncResult;
 import io.vertx.core.Context;
 import io.vertx.core.DeploymentOptions;
+import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.datagram.DatagramSocket;
-import io.vertx.core.datagram.DatagramSocketOptions;
+import io.vertx.core.impl.Utils;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.NetworkOptions;
 import io.vertx.core.streams.WriteStream;
 import io.vertx.test.core.TestUtils;
 import io.vertx.test.core.VertxTestBase;
 import io.vertx.test.netty.TestLoggerFactory;
+import org.junit.Assume;
 import org.junit.Test;
 
 import java.net.InetAddress;
@@ -35,6 +37,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
 
 import static io.vertx.test.core.TestUtils.*;
 
@@ -357,38 +360,68 @@ public void testPause() {
 
   @Test
   public void testMulticastJoinLeave() throws Exception {
+    String iface = NetworkInterface.getByInetAddress(InetAddress.getByName("127.0.0.1")).getName();
+    testMulticastJoinLeave("0.0.0.0", new DatagramSocketOptions(), new DatagramSocketOptions().setMulticastNetworkInterface(iface), (groupAddress, handler) -> {
+      peer1.listenMulticastGroup(groupAddress, iface, null, handler);
+    }, (groupAddress, handler) -> {
+      peer1.unlistenMulticastGroup(groupAddress, iface, null, handler);
+    });
+  }
+
+  @Test
+  public void testMulticastJoinLeaveReuseMulticastNetworkInterface() throws Exception {
+    String iface = NetworkInterface.getByInetAddress(InetAddress.getByName("127.0.0.1")).getName();
+    DatagramSocketOptions options = new DatagramSocketOptions().setMulticastNetworkInterface(iface);
+    testMulticastJoinLeave("0.0.0.0", options, options, (groupAddress, handler) -> {
+      peer1.listenMulticastGroup(groupAddress, handler);
+    }, (groupAddress, handler) -> {
+      peer1.unlistenMulticastGroup(groupAddress, handler);
+    });
+  }
+
+  @Test
+  public void testMulticastJoinLeaveBindOnMulticastGroup() throws Exception {
+    Assume.assumeFalse(Utils.isWindows());
+    String iface = NetworkInterface.getByInetAddress(InetAddress.getByName("127.0.0.1")).getName();
+    DatagramSocketOptions options = new DatagramSocketOptions().setMulticastNetworkInterface(iface);
+    testMulticastJoinLeave("230.0.0.1", options, options, (groupAddress, handler) -> {
+      peer1.listenMulticastGroup(groupAddress, handler);
+    }, (groupAddress, handler) -> {
+      peer1.unlistenMulticastGroup(groupAddress, handler);
+    });
+  }
+
+  private void testMulticastJoinLeave(String bindAddress,
+                                      DatagramSocketOptions options1,
+                                      DatagramSocketOptions options2,
+                                      BiConsumer<String, Handler<AsyncResult<Void>>> join,
+                                      BiConsumer<String, Handler<AsyncResult<Void>>> leave) {
     if (USE_NATIVE_TRANSPORT) {
       return;
     }
-    Buffer buffer = TestUtils.randomBuffer(128);
+    Buffer buffer = Buffer.buffer("HELLO");
     String groupAddress = "230.0.0.1";
-    String iface = NetworkInterface.getByInetAddress(InetAddress.getByName("127.0.0.1")).getName();
     AtomicBoolean received = new AtomicBoolean();
-    peer1 = vertx.createDatagramSocket(new DatagramSocketOptions().setMulticastNetworkInterface(iface));
-    peer2 = vertx.createDatagramSocket(new DatagramSocketOptions().setMulticastNetworkInterface(iface));
+    peer1 = vertx.createDatagramSocket(options1);
+    peer2 = vertx.createDatagramSocket(options2);
 
     peer1.handler(packet -> {
       assertEquals(buffer, packet.data());
       received.set(true);
     });
 
-    peer1.listen(1234, "0.0.0.0", ar1 -> {
-      assertTrue(ar1.succeeded());
-      peer1.listenMulticastGroup(groupAddress, iface, null, ar2 -> {
-        assertTrue(ar2.succeeded());
-        peer2.send(buffer, 1234, groupAddress, ar3 -> {
-          assertTrue(ar3.succeeded());
+    peer1.listen(1234, bindAddress, onSuccess(v1 -> {
+      join.accept(groupAddress, onSuccess(v2 -> {
+        peer2.send(buffer, 1234, groupAddress, onSuccess(ar3 -> {
           // leave group in 1 second so give it enough time to really receive the packet first
           vertx.setTimer(1000, id -> {
-            peer1.unlistenMulticastGroup(groupAddress, iface, null, ar4 -> {
-              assertTrue(ar4.succeeded());
+            leave.accept(groupAddress, onSuccess(ar4 -> {
               AtomicBoolean receivedAfter = new AtomicBoolean();
               peer1.handler(packet -> {
                 // Should not receive any more event as it left the group
                 receivedAfter.set(true);
               });
-              peer2.send(buffer, 1234, groupAddress, ar5 -> {
-                assertTrue(ar5.succeeded());
+              peer2.send(buffer, 1234, groupAddress, onSuccess(v5 -> {
                 // schedule a timer which will check in 1 second if we received a message after the group
                 // was left before
                 vertx.setTimer(1000, id2 -> {
@@ -396,12 +429,30 @@ public void testMulticastJoinLeave() throws Exception {
                   assertTrue(received.get());
                   testComplete();
                 });
-              });
-            });
+              }));
+            }));
           });
-        });
-      });
-    });
+        }));
+      }));
+    }));
+    await();
+  }
+
+  @Test
+  public void testMulticastJoinWithoutNetworkInterface() {
+    peer1 = vertx.createDatagramSocket(new DatagramSocketOptions());
+    peer1.listenMulticastGroup("230.0.0.1", onFailure(err -> {
+      testComplete();
+    }));
+    await();
+  }
+
+  @Test
+  public void testMulticastLeaveWithoutNetworkInterface() {
+    peer1 = vertx.createDatagramSocket(new DatagramSocketOptions());
+    peer1.unlistenMulticastGroup("230.0.0.1", onFailure(err -> {
+      testComplete();
+    }));
     await();
   }
 
@@ -601,4 +652,5 @@ public void start(Promise<Void> startPromise) {
       }));
     }));
     await();
-  }}
+  }
+}