Java 类io.netty.channel.socket.DatagramChannel 实例源码

public void run() {
    try {
        Bootstrap boot = new Bootstrap();
        boot.group(group)
         .channel(NioDatagramChannel.class)
         .handler(new ChannelInitializer<DatagramChannel>() {

            @Override
            protected void initChannel(DatagramChannel ch) throws Exception {
                channel = ch;
                ch.pipeline().addLast(new UdpChannelHandlerServer(TF2UdpServer.this));
            }

         });
        boot.bind(port).sync().channel().closeFuture();
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

public void init(ChannelPipelineFactory factory) throws Exception
{
    id = String.format("%1$020d",
            Math.abs(new Random(System.currentTimeMillis()).nextLong()))
            .getBytes();

    group = new OioEventLoopGroup();
    connectionlessBootstrap = new Bootstrap();
    connectionlessBootstrap.group(group);
    connectionlessBootstrap.option(ChannelOption.SO_BROADCAST, true);
    connectionlessBootstrap.handler(factory);
    connectionlessBootstrap.channel(OioDatagramChannel.class);
    ;
    datagramChannel = (DatagramChannel) connectionlessBootstrap
            .bind(new InetSocketAddress(mcastGroupPort)).sync().channel();
    multicastAddress = new InetSocketAddress(mcastGroupIp, mcastGroupPort);
    NetworkInterface networkInterface = NetworkInterface
            .getByInetAddress(InetAddress.getByName(bindAddress));
    // for (Enumeration nifs = NetworkInterface.getNetworkInterfaces();
    // nifs.hasMoreElements(); )
    datagramChannel.joinGroup(multicastAddress, null);// (NetworkInterface)
                                                        // nifs.nextElement());
    init = true;
    if (debug)
        factory.debug();
}

@Override
public Mono<? extends NettyContext> newHandler(BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>> handler) {
    final BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>>
            targetHandler =
            null == handler ? ChannelOperations.noopHandler() : handler;

    return Mono.create(sink -> {
        Bootstrap b = options.get();
        SocketAddress adr = options.getAddress();
        if(adr == null){
            sink.error(new NullPointerException("Provided UdpServerOptions do not " +
                    "define any address to bind to "));
            return;
        }
        b.localAddress(adr);
        ContextHandler<DatagramChannel> c = doHandler(targetHandler, sink, adr);
        b.handler(c);
        c.setFuture(b.bind());
    });
}

@Override
public Mono<? extends NettyContext> newHandler(BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>> handler) {
    final BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>>
            targetHandler =
            null == handler ? ChannelOperations.noopHandler() : handler;

    return Mono.create(sink -> {
        Bootstrap b = options.get();
        SocketAddress adr = options.getAddress();
        if(adr == null){
            sink.error(new NullPointerException("Provided UdpClientOptions do not " +
                    "define any address to bind to "));
            return;
        }
        b.remoteAddress(adr);
        ContextHandler<DatagramChannel> c = doHandler(targetHandler, sink, adr);
        b.handler(c);
        c.setFuture(b.connect());
    });
}

public void activateOptions() throws InterruptedException {

        Bootstrap b = new Bootstrap();
        b.group(group)
            .channel(NioDatagramChannel.class)
            .handler(new Log4jHandler());
        b.option(ChannelOption.SO_REUSEADDR, true);
        b.option(ChannelOption.IP_MULTICAST_IF, MulticastSettings.getIface());
        b.option(ChannelOption.TCP_NODELAY, true);

        InetSocketAddress addr = new InetSocketAddress(MulticastSettings.getAddressToBind(), port);

        b.localAddress(port).remoteAddress(addr);
        ch = (DatagramChannel) b.bind().sync().channel();

        ch.joinGroup(multicastAddress, MulticastSettings.getIface()).sync();
    }

public void run() throws Exception {
    // Configure the server.
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioDatagramChannel.class)
                .handler(new ChannelInitializer<DatagramChannel>() {
                    @Override
                    public void initChannel(DatagramChannel ch) throws Exception {
                        ch.pipeline().addLast(
                                new LoggingHandler(LogLevel.INFO),
                                new TftpServerHandler());
                    }
                });

        b.bind(port).sync().channel().closeFuture().await();
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(final String[] args) throws Exception {
    final UAS uas = new UAS();
    final EventLoopGroup udpGroup = new NioEventLoopGroup();

    final Bootstrap b = new Bootstrap();
    b.group(udpGroup)
    .channel(NioDatagramChannel.class)
    .handler(new ChannelInitializer<DatagramChannel>() {
        @Override
        protected void initChannel(final DatagramChannel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("decoder", new SipMessageDatagramDecoder());
            pipeline.addLast("encoder", new SipMessageEncoder());
            pipeline.addLast("handler", uas);
        }
    });

    final InetSocketAddress socketAddress = new InetSocketAddress("127.0.0.1", 5060);
    b.bind(socketAddress).sync().channel().closeFuture().await();
}

public void sendBeacon() {
    final BeaconMessage msg = beacon.get();
    channelLock.readLock().lock();
    try {
        for (final DatagramChannel channel : channels.values()) {
            if (channel.isActive()) {
                LOGGER.trace("trying to send on {}", channel);
                channel.writeAndFlush(msg).addListener(
                        new GenericFutureListener<io.netty.util.concurrent.Future<Void>>() {
                            @Override
                            public void operationComplete(io.netty.util.concurrent.Future<Void> future)
                                    throws Exception {
                                LOGGER.trace(future);
                            }
                        });
            }
        }
    } finally {
        channelLock.readLock().unlock();
    }
}

private void doJoin(final ChannelFuture bindFuture, final DatagramChannel channel, final NetworkInterface interf,
        final InetSocketAddress address, final ChannelPromise promise) {
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            LOGGER.entry(bindFuture, channel, interf, address, promise);
            if (bindFuture.isSuccess()) {
                LOGGER.debug("joining group {} with {} using {} and promise {}", address, interf, channel, promise);
                channel.joinGroup(address, interf, promise);
            } else {
                promise.setFailure(bindFuture.cause());
            }
            LOGGER.exit();
        }
    });
}

/**
 * Mock Nettys ChannelHandlerContext with the minimal functions.
 * 
 * @param buf
 *            The buffer to use for decoding
 * @param m2
 *            The message reference to store the result
 * @return The mocked ChannelHandlerContext
 */
@SuppressWarnings("unchecked")
private ChannelHandlerContext mockChannelHandlerContext(final ByteBuf buf,
        final AtomicReference<Message2> m2) {
    ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
    ByteBufAllocator alloc = mock(ByteBufAllocator.class);
    when(ctx.alloc()).thenReturn(alloc);
    when(alloc.ioBuffer()).thenReturn(buf);
    DatagramChannel dc = mock(DatagramChannel.class);
    when(ctx.channel()).thenReturn(dc);
    when(ctx.writeAndFlush(any(), any(ChannelPromise.class))).thenReturn(null);

    Attribute<InetSocketAddress> attr = mock(Attribute.class);
    when(ctx.attr(any(AttributeKey.class))).thenReturn(attr);

    when(ctx.fireChannelRead(any())).then(new Answer<Void>() {
        @Override
        public Void answer(final InvocationOnMock invocation) throws Throwable {
            Object[] args = invocation.getArguments();
            m2.set((Message2) args[0]);
            return null;
        }
    });

    return ctx;
}

private void createServer(Listener listener, EventLoopGroup eventLoopGroup, ChannelFactory<? extends DatagramChannel> channelFactory) {
    this.eventLoopGroup = eventLoopGroup;
    try {
        Bootstrap b = new Bootstrap();
        b.group(eventLoopGroup)//默认是根据机器情况创建Channel,如果机器支持ipv6,则无法使用ipv4的地址加入组播
                .channelFactory(channelFactory)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(getChannelHandler());

        initOptions(b);

        //直接绑定端口，不要指定host，不然收不到组播消息
        b.bind(port).addListener(future -> {
            if (future.isSuccess()) {
                logger.info("udp server start success on:{}", port);
                if (listener != null) listener.onSuccess(port);
            } else {
                logger.error("udp server start failure on:{}", port, future.cause());
                if (listener != null) listener.onFailure(future.cause());
            }
        });
    } catch (Exception e) {
        logger.error("udp server start exception", e);
        if (listener != null) listener.onFailure(e);
        throw new ServiceException("udp server start exception, port=" + port, e);
    }
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    connection.init(ctx.channel(), false);
    if (multicastAddress != null) {
        ((DatagramChannel) ctx.channel()).joinGroup(multicastAddress, networkInterface, null).addListener(future -> {
            if (future.isSuccess()) {
                LOGGER.info("join multicast group success, channel={}, group={}", ctx.channel(), multicastAddress);
            } else {
                LOGGER.error("join multicast group error, channel={}, group={}", ctx.channel(), multicastAddress, future.cause());
            }
        });
    }
    LOGGER.info("init udp channel={}", ctx.channel());
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    connection.close();
    if (multicastAddress != null) {
        ((DatagramChannel) ctx.channel()).leaveGroup(multicastAddress, networkInterface, null).addListener(future -> {
            if (future.isSuccess()) {
                LOGGER.info("leave multicast group success, channel={}, group={}", ctx.channel(), multicastAddress);
            } else {
                LOGGER.error("leave multicast group error, channel={}, group={}", ctx.channel(), multicastAddress, future.cause());
            }
        });
    }
    LOGGER.info("disconnect udp channel={}, connection={}", ctx.channel(), connection);
}

@Override
protected void initChannel(DatagramChannel ch) throws Exception {
    ch.pipeline().addLast("wrapper_encoder", new PocketWrapperEncoder());
    ch.pipeline().addLast("packet_decoder", new PocketDecoder());
    ch.pipeline().addLast("packet_encoder", new PocketEncoder());
    ch.pipeline().addLast("packet_handler", new PocketHandler());
}

protected ChannelHandler setupUdpChannel() {
    return new ChannelInitializer<DatagramChannel>() {

        @Override
        protected void initChannel(DatagramChannel ch) throws Exception {
            ch.pipeline().addLast("logger", new LoggingHandler());
            ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
            ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
            ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
            ch.pipeline().addLast("putDecoder", new UdpDecoder());
            ch.pipeline().addLast(udpWorkerGroup, "putHandler", new TcpPutHandler(dataStore));
        }
    };
}

@Override
protected ChannelHandler setupUdpChannel() {
    return new ChannelInitializer<DatagramChannel>() {

        @Override
        protected void initChannel(DatagramChannel ch) throws Exception {
            ch.pipeline().addLast("logger", new LoggingHandler());
            ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
            ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
            ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
            ch.pipeline().addLast("putDecoder", new UdpDecoder());
            ch.pipeline().addLast("capture", udpRequests);
        }
    };
}

public static Class<? extends DatagramChannel> getDatagramChannelClass(EventLoopGroup eventLoopGroup) {
    if (eventLoopGroup instanceof EpollEventLoopGroup) {
        return EpollDatagramChannel.class;
    } else {
        return NioDatagramChannel.class;
    }
}

/**
 * Create a {@link ContextHandler} for {@link Bootstrap#handler()}
 *
 * @param handler user provided in/out handler
 * @param sink user provided bind handler
 *
 * @return a new {@link ContextHandler}
 */
protected ContextHandler<DatagramChannel> doHandler(BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>> handler,
        MonoSink<NettyContext> sink,
        SocketAddress providedAddress) {
    return ContextHandler.newClientContext(sink,
            options,
            loggingHandler,
            false,
            providedAddress,
            (ch, c, msg) -> UdpOperations.bind(ch, handler, c));
}

/**
 * Create a {@link ContextHandler} for {@link Bootstrap#handler()}
 *
 * @param handler user provided in/out handler
 * @param sink user provided bind handler
 *
 * @return a new {@link ContextHandler}
 */
protected ContextHandler<DatagramChannel> doHandler(BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>> handler,
        MonoSink<NettyContext> sink,
        SocketAddress providedAddress) {
    return ContextHandler.newClientContext(sink,
            options,
            loggingHandler,
            false,
            providedAddress,
            (ch, c, msg) -> UdpOperations.bind(ch, handler, c));
}

@Override
public InetSocketAddress address() {
    Channel c = channel();
    if (c instanceof SocketChannel) {
        return ((SocketChannel) c).remoteAddress();
    }
    if (c instanceof DatagramChannel) {
        InetSocketAddress a = ((DatagramChannel) c).remoteAddress();
        return a != null ? a : ((DatagramChannel)c ).localAddress();
    }
    throw new IllegalStateException("Does not have an InetSocketAddress");
}

/**
 * Return remote address if remote channel {@link NettyContext} otherwise local
 * address if server selector channel.
 *
 * @return remote or local {@link InetSocketAddress}
 */
default InetSocketAddress address(){
    Channel c = channel();
    if (c instanceof SocketChannel) {
        return ((SocketChannel) c).remoteAddress();
    }
    if (c instanceof ServerSocketChannel) {
        return ((ServerSocketChannel) c).localAddress();
    }
    if (c instanceof DatagramChannel) {
        InetSocketAddress a = ((DatagramChannel) c).remoteAddress();
        return a != null ? a : ((DatagramChannel)c ).localAddress();
    }
    throw new IllegalStateException("Does not have an InetSocketAddress");
}

@Override
protected void initChannel(DatagramChannel channel) throws Exception {
    this.channel = channel;
    channel.pipeline()
            .addLast("simpleRaknetHandler", new SimpleRakNetPacketCodec())
            .addLast("raknetDirectPacketHandler", new RakNetDirectPacketHandler(server))
            .addLast("raknetDatagramHandler", new DatagramRakNetPacketCodec(server))
            .addLast("voxelwindDatagramHandler", new RakNetDatagramHandler(server))
            .addLast("tailHandler", new TailHandler());
}

@Override
protected ChannelHandler newChannelInitializer() {
    return new ChannelInitializer<DatagramChannel>() {
        @Override
        protected void initChannel(DatagramChannel ch) throws Exception {
            ChannelPipeline cp = ch.pipeline();
            cp.addLast("ipMatcher", new UdpProtocolHandler(ipMatcher));
            cp.addLast("handler", new LogRecordHandler(board));
        }
    };
}

private void ensureBound() {
    if (!isActive()) {
        throw new IllegalStateException(
                DatagramChannel.class.getName() +
                " must be bound to join a group.");
    }
}

/**
 * Test try to reproduce issue #1335
 */
@Test
public void testBindMultiple() throws Exception {
    DefaultChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    NioEventLoopGroup group = new NioEventLoopGroup();
    try {
        for (int i = 0; i < 100; i++) {
            Bootstrap udpBootstrap = new Bootstrap();
            udpBootstrap.group(group).channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            // Discard
                            ReferenceCountUtil.release(msg);
                        }
                    });
            DatagramChannel datagramChannel = (DatagramChannel) udpBootstrap
                    .bind(new InetSocketAddress(0)).syncUninterruptibly().channel();
            channelGroup.add(datagramChannel);
        }
        Assert.assertEquals(100, channelGroup.size());
    } finally {
        channelGroup.close().sync();
        group.shutdownGracefully().sync();
    }
}

@Override
protected void initChannel(DatagramChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("encoder", new PacketEncoder());
    pipeline.addLast("decoder", new PacketDecoder());
    pipeline.addLast("handler", new PocketServerHandler());
}

private void ensureBound() {
    if (!isActive()) {
        throw new IllegalStateException(
                DatagramChannel.class.getName() +
                " must be bound to join a group.");
    }
}

/**
 * Test try to reproduce issue #1335
 */
@Test
public void testBindMultiple() throws Exception {
    DefaultChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    NioEventLoopGroup group = new NioEventLoopGroup();
    try {
        for (int i = 0; i < 100; i++) {
            Bootstrap udpBootstrap = new Bootstrap();
            udpBootstrap.group(group).channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            // Discard
                            ReferenceCountUtil.release(msg);
                        }
                    });
            DatagramChannel datagramChannel = (DatagramChannel) udpBootstrap
                    .bind(new InetSocketAddress(0)).syncUninterruptibly().channel();
            channelGroup.add(datagramChannel);
        }
        Assert.assertEquals(100, channelGroup.size());
    } finally {
        channelGroup.close().sync();
        group.shutdownGracefully().sync();
    }
}

public DataServerThread() {
    dataBossGroup = new NioEventLoopGroup(DATA_BOSS_THREADS);
    dataWorkerGroup = new NioEventLoopGroup(DATA_WORKER_THREADS);
    try {
        Bootstrap b = new Bootstrap();
        b.group(dataWorkerGroup)
                .channelFactory(new ChannelFactory<Channel>() {
                    @Override
                    public Channel newChannel() {
                        return new NioDatagramChannel(InternetProtocolFamily.IPv4);
                    }

                    @Override
                    public String toString() {
                        return NioDatagramChannel.class.getSimpleName() + ".class";
                    }
                }).handler(new ChannelInitializer<DatagramChannel>() {
                    @Override
                    public void initChannel(DatagramChannel ch) throws Exception {
                        ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
                        ch.pipeline().addLast(new DataMessageHandler());
                        if (LOG.isTraceEnabled()) {
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
                        }
                    }

                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                        LOG.error("Cannot initialize data server.", cause);
                    }
                });

        // Bind and start to accept incoming connections.
        f = b.bind(getIp(), getDataPort()).sync();
    } catch (InterruptedException ex) {
        LOG.error("Message data interrupted.", ex);
    }
}

protected static Channel createAcceptorChannel(
          final ChannelType channelType,
          final InetSocketAddress localAddress,
          final ServerUDPHandler serverHandler
  ) {
      final Bootstrap serverBootstrap = ServerUDPBootstrapFactory.createServerBootstrap(channelType);

      serverBootstrap
              .option(ChannelOption.SO_REUSEADDR, false)


            .handler(new ChannelInitializer<DatagramChannel>() {
                  @Override
                  protected void initChannel(final DatagramChannel ch) throws Exception {
                      final ChannelPipeline pipeline = ch.pipeline();
                     //pipeline.addLast("readTimeoutHandler", new ReadTimeoutHandler(60));
                      pipeline.addLast("messageDecoder", serverHandler);
                      //pipeline.addLast("handler", serverHandler);
                  }
              });

      try {

          ChannelFuture channelFuture = serverBootstrap.bind(
                  new InetSocketAddress(localAddress.getPort())).sync();

          //channelFuture.channel().closeFuture().awaitUninterruptibly();//.awaitUninterruptibly();
          channelFuture.awaitUninterruptibly();
          if (channelFuture.isSuccess()) {
              return channelFuture.channel();

          } else {

          }
      } catch (InterruptedException e) {

      }
return null;

  }

@Override
protected void doStart() {
  nioEventLoopGroup.next().execute(() -> {

    Bootstrap bootstrap = new Bootstrap();
    try {
      bootstrap.group(nioEventLoopGroup)
          .channel(NioDatagramChannel.class)
          .option(ChannelOption.SO_BROADCAST, true)
          .option(ChannelOption.SO_REUSEADDR, true)
          .handler(new ChannelInitializer<DatagramChannel>() {
            @Override
            protected void initChannel(DatagramChannel ch) throws Exception {
              ChannelPipeline p = ch.pipeline();
              p.addLast("protostuffDecoder",
                  new UdpProtostuffDecoder<>(EventLogEntry.getSchema(), false));
              p.addLast("logger",
                  new MsgHandler());
            }
          });

      bootstrap.bind(port).addListener(new GenericFutureListener<ChannelFuture>() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
          channel = future.channel();
        }
      });


      notifyStarted();
    } catch (Throwable t) {
      notifyFailed(t);
    }
  });
}

private void ensureBound() {
    if (!isActive()) {
        throw new IllegalStateException(
                DatagramChannel.class.getName() +
                " must be bound to join a group.");
    }
}

/**
 * Test try to reproduce issue #1335
 */
@Test
public void testBindMultiple() throws Exception {
    DefaultChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    NioEventLoopGroup group = new NioEventLoopGroup();
    try {
        for (int i = 0; i < 100; i++) {
            Bootstrap udpBootstrap = new Bootstrap();
            udpBootstrap.group(group).channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            // Discard
                            ReferenceCountUtil.release(msg);
                        }
                    });
            DatagramChannel datagramChannel = (DatagramChannel) udpBootstrap
                    .bind(new InetSocketAddress(0)).syncUninterruptibly().channel();
            channelGroup.add(datagramChannel);
        }
        Assert.assertEquals(100, channelGroup.size());
    } finally {
        channelGroup.close().sync();
        group.shutdownGracefully().sync();
    }
}

public NettySession(InetSocketAddress remotePublicSocketAddress,
        InetSocketAddress localSocketAddress, Bootstrap bootstrap, DatagramChannel channel) {
    this.remotePublicSocketAddress = remotePublicSocketAddress;
    this.localSocketAddress = localSocketAddress;
    this.bootstrap = bootstrap;
    this.channel = channel;
}

private Bootstrap createUDPListeningPoint(final SimpleChannelInboundHandler<SipMessageEvent> handler) {
    final Bootstrap b = new Bootstrap();
    b.group(this.udpGroup)
    .channel(NioDatagramChannel.class)
    .handler(new ChannelInitializer<DatagramChannel>() {
        @Override
        protected void initChannel(final DatagramChannel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("decoder", new SipMessageDatagramDecoder());
            pipeline.addLast("encoder", new SipMessageEncoder());
            pipeline.addLast("handler", handler);
        }
    });
    return b;
}

/**
 * Creates a new UDPMulticastBeacon.
 * 
 * @param factory
 *            a ChannelFactory
 * @param group
 *            the EventLoopGroup to use for channels and the timer
 * @param executor
 *            the executor for application code and a timer for regularly sending the beacon
 * @param moduleID
 *            the ModuleID to announce
 * @param interval
 *            the interval at which to send beacons
 * @param unit
 *            the unit for interval
 */
public UDPMulticastBeacon(final ChannelFactory<? extends DatagramChannel> factory, final EventLoopGroup group,
        final ScheduledExecutorService executor, final ModuleID moduleID, final long interval, final TimeUnit unit) {
    beacon =
            new AtomicReference<BeaconMessage>(new BeaconMessage(moduleID,
                    Collections.<InetSocketAddress> emptyList()));

    executor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            sendBeacon();
        }
    }, 0, interval, unit);

    final MessageAdapter messageAdapter = new MessageAdapter();
    messageAdapter.addMessageType(BeaconMessage.class);
    final GsonCodec gsonCodec = new GsonCodec(Message.class);
    gsonCodec.registerTypeAdapter(Message.class, messageAdapter);
    gsonCodec.registerTypeAdapter(InetSocketAddress.class, new InetSocketAddressAdapter());
    gsonCodec.registerTypeAdapter(BeaconMessage.class, new BeaconMessageDeserializer());

    this.channelFactory = new UDPMulticastChannelFactory(factory, group, new ChannelInitializer<DatagramChannel>() {
        private final DatagramPacketWrapper datagramPacketWrapper = new DatagramPacketWrapper();
        private final StringEncoder stringEncoder = new StringEncoder();
        private final StringDecoder stringDecoder = new StringDecoder();
        private final ChannelHandler beaconHandler = new BeaconHandler();

        @Override
        protected void initChannel(final DatagramChannel channel) {
            channel.pipeline().addLast(datagramPacketWrapper).addLast(stringEncoder).addLast(stringDecoder)
                    .addLast(gsonCodec).addLast(beaconHandler);

            // Move TARGET_ADDRESS from channel context to handler context
            channel.pipeline().context(DatagramPacketWrapper.class).attr(DatagramPacketWrapper.TARGET_ADDRESS)
                    .set(channel.attr(DatagramPacketWrapper.TARGET_ADDRESS).getAndRemove());
        }
    });
}

private void doBind(final ChannelFuture regFuture, final DatagramChannel channel, final NetworkInterface interf,
        final InetSocketAddress address, final ChannelPromise bindPromise, final ChannelPromise joinPromise) {
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            LOGGER.entry(regFuture, channel, address, bindPromise, joinPromise);
            if (regFuture.isSuccess()) {
                LOGGER.debug("trying to bind {} to {} with promise {}", channel, address, bindPromise);
                channel.config().setNetworkInterface(interf);
                channel.config().setReuseAddress(true);
                channel.bind(new InetSocketAddress(address.getPort()), bindPromise);
            } else {
                bindPromise.setFailure(regFuture.cause());
            }
            if (bindPromise.isDone()) {
                LOGGER.debug("doing join now");
                doJoin(bindPromise, channel, interf, address, joinPromise);
            } else {
                LOGGER.debug("doing join later");
                bindPromise.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(final ChannelFuture future) throws Exception {
                        doJoin(bindPromise, channel, interf, address, joinPromise);
                    }
                });
            }
            LOGGER.exit();
        }
    });
}

private boolean bindUdpPort(final InetAddress addr, final int port) {

        EventLoopGroup group = new NioEventLoopGroup();
        bootstrapUDP = new Bootstrap();
        bootstrapUDP.group(group).channel(NioDatagramChannel.class)
                .handler(new DatagramHandler(this, Transport.UDP));

        bootstrapUDP.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(1500, 1500, RECV_BUFFER_SIZE));
        bootstrapUDP.option(ChannelOption.SO_RCVBUF, RECV_BUFFER_SIZE);
        bootstrapUDP.option(ChannelOption.SO_SNDBUF, SEND_BUFFER_SIZE);
        // bootstrap.setOption("trafficClass", trafficClass);
        // bootstrap.setOption("soTimeout", soTimeout);
        // bootstrap.setOption("broadcast", broadcast);
        bootstrapUDP.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIMEOUT_MS);
        bootstrapUDP.option(ChannelOption.SO_REUSEADDR, true);

        try {
            InetSocketAddress iAddr = new InetSocketAddress(addr, port);
            udpChannel = (DatagramChannel) bootstrapUDP.bind(iAddr).sync().channel();

            //addLocalSocket(iAddr, c);
            logger.info("Successfully bound to ip:port {}:{}", addr, port);
        } catch (InterruptedException e) {
            logger.error("Problem when trying to bind to {}:{}", addr.getHostAddress(), port);
            return false;
        }

        return true;
    }

@Override
protected void initChannel(DatagramChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("hello", new HelloHandler());
    pipeline.addLast("playerInfo", new PlayerInfoHandler());
}

@Override
public Class<? extends DatagramChannel> onDatagramChannel(EventLoopGroup group) {
    return defaultLoops.onDatagramChannel(group);
}