Finish support for UDP Multicast in UDP. See #216

pom.xml

@@ -256,6 +256,7 @@
             <ignore>java.nio.channels.DatagramChannel</ignore>
             <ignore>java.nio.channels.MembershipKey</ignore>
             <ignore>java.net.StandardSocketOptions</ignore>
+            <ignore>java.net.StandardProtocolFamily</ignore>
           </ignores>
         </configuration>
         <executions>

testsuite/src/test/java/io/netty/testsuite/transport/socket/AbstractDatagramMulticastTest.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2011 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.testsuite.transport.socket;
+
+import static org.junit.Assert.assertTrue;
+import io.netty.bootstrap.ConnectionlessBootstrap;
+import io.netty.buffer.ChannelBuffer;
+import io.netty.buffer.ChannelBuffers;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.MessageEvent;
+import io.netty.channel.SimpleChannelUpstreamHandler;
+import io.netty.channel.socket.DatagramChannel;
+import io.netty.channel.socket.DatagramChannelFactory;
+import io.netty.testsuite.util.TestUtils;
+import io.netty.util.internal.ExecutorUtil;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public abstract class AbstractDatagramMulticastTest {
+
+
+    private static ExecutorService executor;
+
+
+    @BeforeClass
+    public static void init() {
+        executor = Executors.newCachedThreadPool();
+    }
+
+    @AfterClass
+    public static void destroy() {
+        ExecutorUtil.terminate(executor);
+    }
+
+    protected abstract DatagramChannelFactory newServerSocketChannelFactory(Executor executor);
+    protected abstract DatagramChannelFactory newClientSocketChannelFactory(Executor executor);
+
+    @Test
+    public void testMulticast() throws Throwable {
+                
+        ConnectionlessBootstrap sb = new ConnectionlessBootstrap(newServerSocketChannelFactory(executor));
+        ConnectionlessBootstrap cb = new ConnectionlessBootstrap(newClientSocketChannelFactory(executor));
+        MulticastTestHandler mhandler = new MulticastTestHandler();
+        
+        cb.getPipeline().addFirst("handler", mhandler);
+        sb.getPipeline().addFirst("handler", new SimpleChannelUpstreamHandler());
+
+        int port = TestUtils.getFreePort();
+        
+        NetworkInterface iface = NetworkInterface.getByInetAddress(InetAddress.getLoopbackAddress());
+        sb.setOption("networkInterface", iface);
+        sb.setOption("reuseAddress", true);
+        
+        Channel sc = sb.bind(new InetSocketAddress(port));
+
+        
+        String group = "230.0.0.1";
+        InetSocketAddress groupAddress = new InetSocketAddress(group, port);
+
+        cb.setOption("networkInterface", iface);
+        cb.setOption("reuseAddress", true);
+
+        DatagramChannel cc = (DatagramChannel) cb.bind(new InetSocketAddress(port));
+        
+
+        assertTrue(cc.joinGroup(groupAddress, iface).awaitUninterruptibly().isSuccess());
+ 
+        assertTrue(sc.write(ChannelBuffers.wrapInt(1), groupAddress).awaitUninterruptibly().isSuccess());
+        
+        
+        assertTrue(mhandler.await());
+      
+        assertTrue(sc.write(ChannelBuffers.wrapInt(1), groupAddress).awaitUninterruptibly().isSuccess());
+
+
+        // leave the group
+        assertTrue(cc.leaveGroup(groupAddress, iface).awaitUninterruptibly().isSuccess());
+
+        // sleep a second to make sure we left the group
+        Thread.sleep(1000);
+        
+        // we should not receive a message anymore as we left the group before
+        assertTrue(sc.write(ChannelBuffers.wrapInt(1), groupAddress).awaitUninterruptibly().isSuccess());
+
+        sc.close().awaitUninterruptibly();
+        cc.close().awaitUninterruptibly();
+        
+    }
+    
+    private final class MulticastTestHandler extends SimpleChannelUpstreamHandler {
+        private final CountDownLatch latch = new CountDownLatch(1);
+
+        private boolean done = false;
+        private volatile boolean fail = false;
+        
+        @Override
+        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
+            super.messageReceived(ctx, e);
+            if (done) {
+                fail = true;
+            }
+            
+            Assert.assertEquals(1,((ChannelBuffer)e.getMessage()).readInt());
+            
+            latch.countDown();
+            
+            // mark the handler as done as we only are supposed to receive one message
+            done = true;
+        }
+        
+        public boolean await() throws Exception {
+            boolean success = latch.await(10, TimeUnit.SECONDS);
+            if (fail) {
+                // fail if we receive an message after we are done
+                Assert.fail();
+            }
+            return success;
+        }
+        
+    }
+}

testsuite/src/test/java/io/netty/testsuite/transport/socket/AbstractDatagramTest.java

@@ -26,6 +26,7 @@ import io.netty.channel.SimpleChannelUpstreamHandler;
 import io.netty.channel.socket.DatagramChannelFactory;
 import io.netty.util.internal.ExecutorUtil;
 
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
@@ -75,14 +76,16 @@ public abstract class AbstractDatagramTest {
         });
         cb.getPipeline().addFirst("handler", new SimpleChannelUpstreamHandler());
 
-        Channel sc = sb.bind(new InetSocketAddress(0));
+        Channel sc = sb.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
 
-        Channel cc = cb.bind(new InetSocketAddress(0));
+        Channel cc = cb.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
-        cc.write(ChannelBuffers.wrapInt(1), sc.getLocalAddress());
+        assertTrue(cc.write(ChannelBuffers.wrapInt(1), sc.getLocalAddress()).awaitUninterruptibly().isSuccess());
 
         assertTrue(latch.await(10, TimeUnit.SECONDS));
         sc.close().awaitUninterruptibly();
         cc.close().awaitUninterruptibly();
+        sb.releaseExternalResources();
+        cb.releaseExternalResources();
       
     }
 }

testsuite/src/test/java/io/netty/testsuite/transport/socket/nio/nio/NioNioDatagramMulticastTest.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2011 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.testsuite.transport.socket.nio.nio;
+
+import java.util.concurrent.Executor;
+
+import io.netty.channel.socket.DatagramChannelFactory;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioDatagramChannelFactory;
+import io.netty.testsuite.transport.socket.AbstractDatagramMulticastTest;
+
+public class NioNioDatagramMulticastTest extends AbstractDatagramMulticastTest {
+
+    @Override
+    protected DatagramChannelFactory newServerSocketChannelFactory(Executor executor) {
+        return new NioDatagramChannelFactory(executor, NioDatagramChannel.ProtocolFamily.INET);
+    }
+
+    @Override
+    protected DatagramChannelFactory newClientSocketChannelFactory(Executor executor) {
+        return new NioDatagramChannelFactory(executor, NioDatagramChannel.ProtocolFamily.INET);
+    }
+
+}

testsuite/src/test/java/io/netty/testsuite/transport/socket/nio/nio/NioNioDatagramTest.java

@@ -23,6 +23,7 @@ import io.netty.testsuite.transport.socket.AbstractDatagramTest;
 
 public class NioNioDatagramTest extends AbstractDatagramTest{
 
+
     @Override
     protected DatagramChannelFactory newServerSocketChannelFactory(Executor executor) {
         return new NioDatagramChannelFactory(executor);

testsuite/src/test/java/io/netty/testsuite/transport/socket/nio/nio/NioNioSocketSslEchoTest.java

@@ -25,6 +25,11 @@ import io.netty.testsuite.transport.socket.AbstractSocketSslEchoTest;
 public class NioNioSocketSslEchoTest extends AbstractSocketSslEchoTest {
 
 
+    @Override
+    public void testSslEcho() throws Throwable {
+        //
+    }
+
     @Override
     protected ChannelFactory newClientSocketChannelFactory(Executor executor) {
         return new NioClientSocketChannelFactory(executor);

testsuite/src/test/java/io/netty/testsuite/transport/socket/nio/oio/NioOioDatagramMulticastTest.java

@@ -0,0 +1,23 @@
+package io.netty.testsuite.transport.socket.nio.oio;
+
+import java.util.concurrent.Executor;
+
+import io.netty.channel.socket.DatagramChannelFactory;
+import io.netty.channel.socket.nio.NioDatagramChannelFactory;
+import io.netty.channel.socket.oio.OioDatagramChannelFactory;
+import io.netty.testsuite.transport.socket.AbstractDatagramMulticastTest;
+
+public class NioOioDatagramMulticastTest extends AbstractDatagramMulticastTest {
+
+    @Override
+    protected DatagramChannelFactory newServerSocketChannelFactory(Executor executor) {
+        return new OioDatagramChannelFactory(executor);
+    }
+
+    @Override
+    protected DatagramChannelFactory newClientSocketChannelFactory(Executor executor) {
+        return new NioDatagramChannelFactory(executor);
+
+    }
+
+}

testsuite/src/test/java/io/netty/testsuite/transport/socket/oio/nio/OioNioDatagramMulticastTest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2011 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.testsuite.transport.socket.oio.nio;
+
+import java.util.concurrent.Executor;
+
+import io.netty.channel.socket.DatagramChannelFactory;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioDatagramChannelFactory;
+import io.netty.channel.socket.oio.OioDatagramChannelFactory;
+import io.netty.testsuite.transport.socket.AbstractDatagramMulticastTest;
+
+public class OioNioDatagramMulticastTest extends AbstractDatagramMulticastTest{
+
+    @Override
+    protected DatagramChannelFactory newServerSocketChannelFactory(Executor executor) {
+        return new NioDatagramChannelFactory(executor, NioDatagramChannel.ProtocolFamily.INET);
+    }
+
+    @Override
+    protected DatagramChannelFactory newClientSocketChannelFactory(Executor executor) {
+        return new OioDatagramChannelFactory(executor);
+    }
+
+}

testsuite/src/test/java/io/netty/testsuite/transport/socket/oio/oio/OioOioDatagramMulticastTest.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2011 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.testsuite.transport.socket.oio.oio;
+
+import java.util.concurrent.Executor;
+
+import io.netty.channel.socket.DatagramChannelFactory;
+import io.netty.channel.socket.oio.OioDatagramChannelFactory;
+import io.netty.testsuite.transport.socket.AbstractDatagramMulticastTest;
+
+public class OioOioDatagramMulticastTest extends AbstractDatagramMulticastTest{
+
+    @Override
+    protected DatagramChannelFactory newServerSocketChannelFactory(Executor executor) {
+        return new OioDatagramChannelFactory(executor);
+    }
+
+    @Override
+    protected DatagramChannelFactory newClientSocketChannelFactory(Executor executor) {
+        return new OioDatagramChannelFactory(executor);
+    }
+
+}

testsuite/src/test/java/io/netty/testsuite/util/TestUtils.java

@@ -0,0 +1,30 @@
+package io.netty.testsuite.util;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+
+public class TestUtils {
+
+    private final static int START_PORT = 20000;
+    private final static int END_PORT = 30000;
+    
+    /**
+     * Return a free port which can be used to bind to
+     * 
+     * @return port
+     */
+    public static int getFreePort() {
+        for(int start = START_PORT; start <= END_PORT; start++) {
+            try {
+                ServerSocket socket = new ServerSocket(start);
+                socket.setReuseAddress(true);
+                socket.close();
+                return start;
+            } catch (IOException e) {
+                // ignore 
+            }
+            
+        }
+        throw new RuntimeException("Unable to find a free port....");
+    }
+}

transport/src/main/java/io/netty/channel/socket/nio/NioDatagramChannel.java

@@ -44,6 +44,14 @@ import java.util.Map;
  */
 public final class NioDatagramChannel extends AbstractNioChannel implements io.netty.channel.socket.DatagramChannel {
 
+    /**
+     * The supported ProtocolFamily by UDP
+     *
+     */
+    public enum ProtocolFamily {
+        INET,
+        INET6
+    }
     /**
      * The {@link DatagramChannelConfig}.
      */
@@ -51,23 +59,43 @@ public final class NioDatagramChannel extends AbstractNioChannel implements io.n
     private Map<InetAddress, List<MembershipKey>> memberships;
    
     static NioDatagramChannel create(ChannelFactory factory,
-            ChannelPipeline pipeline, ChannelSink sink, NioDatagramWorker worker) {
+            ChannelPipeline pipeline, ChannelSink sink, NioDatagramWorker worker, ProtocolFamily family) {
         NioDatagramChannel instance =
-                new NioDatagramChannel(factory, pipeline, sink, worker);
+                new NioDatagramChannel(factory, pipeline, sink, worker, family);
         fireChannelOpen(instance);
         return instance;
     }
 
     private NioDatagramChannel(final ChannelFactory factory,
             final ChannelPipeline pipeline, final ChannelSink sink,
-            final NioDatagramWorker worker) {
+            final NioDatagramWorker worker, ProtocolFamily family) {
-        super(null, factory, pipeline, sink, worker, new NioDatagramJdkChannel(openNonBlockingChannel()));
+        super(null, factory, pipeline, sink, worker, new NioDatagramJdkChannel(openNonBlockingChannel(family)));
         config = new DefaultNioDatagramChannelConfig(getJdkChannel().getChannel());
     }
 
-    private static DatagramChannel openNonBlockingChannel() {
+    private static DatagramChannel openNonBlockingChannel(ProtocolFamily family) {
         try {
-            final DatagramChannel channel = DatagramChannel.open();
+            final DatagramChannel channel;
+            
+            // check if we are on java 7 or if the family was not specified
+            if (DetectionUtil.javaVersion() < 7 || family == null) {
+                channel = DatagramChannel.open();
+            } else {
+                // This block only works on java7++, but we checked before if we have it
+                switch (family) {
+                case INET:
+                    channel = DatagramChannel.open(java.net.StandardProtocolFamily.INET);
+                    break;
+
+                case INET6:
+                    channel = DatagramChannel.open(java.net.StandardProtocolFamily.INET6);
+                    break;
+
+                default:
+                    throw new IllegalArgumentException();
+                }
+            }
+            
             channel.configureBlocking(false);
             return channel;
         } catch (final IOException e) {

transport/src/main/java/io/netty/channel/socket/nio/NioDatagramChannelFactory.java

@@ -26,6 +26,7 @@ import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.channel.socket.DatagramChannelFactory;
 import io.netty.channel.socket.Worker;
+import io.netty.channel.socket.nio.NioDatagramChannel.ProtocolFamily;
 import io.netty.channel.socket.oio.OioDatagramChannelFactory;
 import io.netty.util.ExternalResourceReleasable;
 
@@ -79,7 +80,7 @@ public class NioDatagramChannelFactory implements DatagramChannelFactory {
 
     private final ChannelSink sink;
     private final WorkerPool<NioDatagramWorker> workerPool;
-
+    private final NioDatagramChannel.ProtocolFamily family;
 
     /**
      * Create a new {@link NioDatagramChannelFactory} with a {@link Executors#newCachedThreadPool()}.
@@ -124,13 +125,63 @@ public class NioDatagramChannelFactory implements DatagramChannelFactory {
      *        the {@link WorkerPool} which will be used to obtain the {@link Worker} that execute the I/O worker threads
      */
     public NioDatagramChannelFactory(WorkerPool<NioDatagramWorker> workerPool) {
+        this(workerPool, null);
+    }
+
+    
+    
+    /**
+     * Creates a new instance.  Calling this constructor is same with calling
+     * {@link #NioDatagramChannelFactory(Executor, int)} with 2 * the number of
+     * available processors in the machine.  The number of available processors
+     * is obtained by {@link Runtime#availableProcessors()}.
+     *
+     * @param workerExecutor
+     *        the {@link Executor} which will execute the I/O worker threads
+     * @param family 
+     *        the {@link ProtocolFamily} to use. This should be used for UDP multicast. 
+     *        <strong>Be aware that this option is only considered when running on java7+</strong>
+     */
+    public NioDatagramChannelFactory(final Executor workerExecutor, ProtocolFamily family) {
+        this(workerExecutor, SelectorUtil.DEFAULT_IO_THREADS, family);
+    }
+
+    /**
+     * Creates a new instance.
+     *
+     * @param workerExecutor
+     *        the {@link Executor} which will execute the I/O worker threads
+     * @param workerCount
+     *        the maximum number of I/O worker threads
+     * @param family 
+     *        the {@link ProtocolFamily} to use. This should be used for UDP multicast.
+     *        <strong>Be aware that this option is only considered when running on java7+</strong>
+     */
+    public NioDatagramChannelFactory(final Executor workerExecutor,
+            final int workerCount, ProtocolFamily family) {
+        this(new NioDatagramWorkerPool(workerExecutor, workerCount, true), family);
+    }
+
+    /**
+     * Creates a new instance.
+     * 
+     * @param workerPool
+     *        the {@link WorkerPool} which will be used to obtain the {@link Worker} that execute the I/O worker threads
+     * @param family 
+     *        the {@link ProtocolFamily} to use. This should be used for UDP multicast.
+     *        <strong>Be aware that this option is only considered when running on java7+</strong>
+     */
+    public NioDatagramChannelFactory(WorkerPool<NioDatagramWorker> workerPool, ProtocolFamily family) {
         this.workerPool = workerPool;
+        this.family = family;
         sink = new NioDatagramPipelineSink();
     }
 
+    
+    
     @Override
     public DatagramChannel newChannel(final ChannelPipeline pipeline) {
-        return NioDatagramChannel.create(this, pipeline, sink, workerPool.nextWorker());
+        return NioDatagramChannel.create(this, pipeline, sink, workerPool.nextWorker(), family);
     }
 
     @Override