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

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof DatagramPacket) {
        // Get packet and sender data
        DatagramPacket datagram = (DatagramPacket) msg;
        InetSocketAddress sender = datagram.sender();
        RakNetPacket packet = new RakNetPacket(datagram);

        // If an exception happens it's because of this address
        this.causeAddress = sender;

        // Handle the packet and release the buffer
        client.handleMessage(packet, sender);
        datagram.content().readerIndex(0); // Reset position
        client.getListener().handleNettyMessage(datagram.content(), sender);
        datagram.content().release(); // No longer needed

        // No exceptions occurred, release the suspect
        this.causeAddress = null;
    }
}

private SimpleChannelInboundHandler<DatagramPacket> createListenerHandler(SeedNode thisNode, ByteBuf seedNodeInfo) {
    return new SimpleChannelInboundHandler<DatagramPacket>() {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
            ByteBuf buf = msg.content();

            if (buf.readableBytes() > 4 && buf.readInt() == Utils.MAGIC_BYTES) {
                MessageTYpe msgType = MessageTYpe.values()[buf.readByte()];

                if (msgType == MessageTYpe.DISCOVERY) {
                    String cluster = decodeUtf(buf);
                    InetSocketAddress address = decodeAddress(buf);

                    if (thisNode.cluster().equals(cluster) && !address.equals(thisNode.address())) {
                        onDiscoveryMessage(address);

                        DatagramPacket response = new DatagramPacket(seedNodeInfo.copy(), msg.sender());

                        ctx.writeAndFlush(response);
                    }
                }
            }
        }
    };
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
  // Creates the client request
  RtMessage msg = client.createRequest();

  // Encodes the request for network transmission
  ByteBuf encodedMsg = RtWire.toWire(msg);

  // Sends the request to the Roughtime server
  ctx.writeAndFlush(new DatagramPacket(encodedMsg, addr))
      .addListener(fut -> {
        if (!fut.isSuccess()) {
          System.out.println("Send failed " + fut.cause().getMessage());
        }
      });
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
    ByteBuf buffer = msg.content();

    int seq = buffer.readUnsignedShort();
    int msgid = buffer.readByte();

    IMessage message = TF2weapons.network.messages[msgid].newInstance();
    //buffer.discardReadBytes();
    message.fromBytes(buffer);
    IMessageHandler<IMessage, IMessage> handler = TF2weapons.network.handlerList.get(message.getClass());
    if(constr == null) {
        constr =MessageContext.class.getDeclaredConstructor(INetHandler.class, Side.class);
        constr.setAccessible(true);
    }
    MessageContext context = constr.newInstance(Minecraft.getMinecraft().player.connection, Side.CLIENT); 
    handler.onMessage(message, context);
    System.out.println("PacketFrom: "+msg.sender().getAddress()+ " "+msg.sender().getPort()+" ");

}

public void sendToAll(IMessage message)
{

    if(useUdp && discriminators.containsKey(message.getClass())) {
        for (EntityPlayer player : TF2weapons.server.getPlayerList().getPlayers()) {
            InetSocketAddress address=TF2weapons.udpServer.outboundTargets.get(player.getCapability(TF2weapons.PLAYER_CAP, null).udpServerId);
            if(address != null) {
                ByteBuf buffer = Unpooled.buffer();
                buffer.writeShort(0);
                buffer.writeByte(discriminators.get(message.getClass()));
                message.toBytes(buffer);
                DatagramPacket packet = new DatagramPacket(buffer, address);
                TF2weapons.udpServer.channel.writeAndFlush(packet);
            }
            else {
                super.sendTo(message, (EntityPlayerMP) player);
            }
        }
    }
    else {
        super.sendToAll(message);
    }
}

public void sendTo(IMessage message, EntityPlayerMP player)
{
    if(useUdp && discriminators.containsKey(message.getClass())) {
        InetSocketAddress address=TF2weapons.udpServer.outboundTargets.get(player.getCapability(TF2weapons.PLAYER_CAP, null).udpServerId);
        if (address != null) {
        ByteBuf buffer = Unpooled.buffer();
        buffer.writeShort(0);
        buffer.writeByte(discriminators.get(message.getClass()));
        message.toBytes(buffer);
        DatagramPacket packet = new DatagramPacket(buffer, address);
        TF2weapons.udpServer.channel.writeAndFlush(packet);
        }
    }
    else {
        super.sendTo(message, player);
    }
}

public void sendToServer(IMessage message)
{
    if(useUdp && TF2UdpClient.instance != null && discriminators.containsKey(message.getClass())) {
        InetSocketAddress address=TF2UdpClient.instance.address;
        //System.out.println("Addr: "+address);
        if (address != null) {
        ByteBuf buffer = Unpooled.buffer();
        buffer.writeShort(TF2UdpClient.playerId);
        buffer.writeShort(0);
        buffer.writeByte(discriminators.get(message.getClass()));
        message.toBytes(buffer);
        DatagramPacket packet = new DatagramPacket(buffer, address);
        TF2UdpClient.instance.channel.writeAndFlush(packet);
        }
    }
    else {
        super.sendToServer(message);
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup loopGroup = new NioEventLoopGroup();
    try{
        Bootstrap b = new Bootstrap();
        b.group(loopGroup).channel(NioDatagramChannel.class)
        .option(ChannelOption.SO_BROADCAST, true)
        .handler(new ChannelProverbClientHandler());
        Channel ch = b.bind(0).sync().channel();
        //向网段内所有的机器广播UDP消息
        ch.writeAndFlush(
                new DatagramPacket(Unpooled.copiedBuffer("谚语字典查询？", CharsetUtil.UTF_8),
                new InetSocketAddress("255.255.255.255", port))).sync();
        //客户端等待15s接受服务端的应答消息，然后退出释放资源
        if(!ch.closeFuture().await(15000)){
            System.out.println("查询超时");
        }
    }finally{
        loopGroup.shutdownGracefully();
    }

}

public void send(ByteBuf msg) throws Exception
{

    byte[] arr = msg.array();
    byte[] buf = new byte[arr.length + id.length];
    System.arraycopy(id, 0, buf, 0, id.length);
    System.arraycopy(arr, 0, buf, id.length, arr.length);

    ByteBuf bbuf = Unpooled.wrappedBuffer(buf);

    if (debug && logger != null)
        logger.info("discovery send " + new String(bbuf.array()));

    datagramChannel.writeAndFlush(
            new DatagramPacket(bbuf, multicastAddress)).sync();

    // datagramChannel.writeAndFlush(buf, multicastAddress);
}

@Override
protected void encode(ChannelHandlerContext ctx, PacketWrapper msg, List<Object> out) throws Exception {
    ctx.attr(PipelineUtil.ADDRESS_ATTRIBUTE).set(msg.getRecipient());
    if (msg.getPacket().getType().isRaw()) {
        ctx.write(msg.getPacket());
        return;
    }
    PacketRaknetOutCustomPacket.writeMany(ctx, msg).forEach(outgoing -> {
        ByteBuf buffer = outgoing.getBuffer();
        if (PocketEncoder.dump) {
            PocketServer.getInstance().getLogger().debug("Encoded: {}", ByteBufUtil.hexDump(buffer).toUpperCase());
        }
        Consumer<Session> receipt = msg.getAckReceipt();
        if (receipt != null) {
            Session session = PocketServer.getInstance().getSessions().get(msg.getRecipient());
            session.addAckReceipt(outgoing.getSequenceNumber(), receipt);
        }
        out.add(new DatagramPacket(buffer, msg.getRecipient()));
    });
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
    channel = ctx.channel();
    sender = msg.sender();

    ByteBuf content = msg.content();

    int methodID = content.readInt();
    int invID = content.readInt();
    int argCount = content.readInt();
    Object[] args = new Object[argCount];
    for(int i = 0; i < argCount; i++) {
        Class<?> clazz = Class.forName(readString(content));
        args[i] = gson.fromJson(readString(content), clazz);
    }

    provider.invoke(methodID, invID, args);
}

@Override
public void sendInvocationResult(int invocationID, Object result) {
    ByteBuf buf = Unpooled.buffer();
    buf.writeInt(invocationID);
    String className = result == null ? Void.class.getCanonicalName() : result.getClass().getCanonicalName();
    writeString(buf, className);
    if(result instanceof ByteBuf) {
        ByteBuf res = (ByteBuf) result;
        buf.writeBytes(res);
    } else {
        String json = gson.toJson(result);
        writeString(buf, json);
    }

    DatagramPacket msg = new DatagramPacket(buf, sender);
    channel.writeAndFlush(msg);
}

@Override
protected void decode(ChannelHandlerContext ctx, DatagramPacket msg, List<Object> out) throws Exception {
    //Create our response packet from the datagram we received
    final MasterServerResponsePacket packet = builder.construct(msg.content());
    if (packet != null) {
        final MasterServerResponse response = new MasterServerResponse();
        if (response != null) {
            response.setSender(msg.sender());
            response.setRecipient(msg.recipient());
            response.setResponsePacket(packet);
            log.debug("Receiving Data '{}' from '{}' using Channel Id: {}", response.getClass().getSimpleName(), ctx.channel().remoteAddress(), ctx.channel().id());
            //Pass the message back to the messenger
            responseCallback.accept(response, null);
            return;
        }
    }
    throw new IllegalStateException("No response packet found for the incoming datagram");
}

@Override
@SuppressWarnings("unchecked")
protected void decode(ChannelHandlerContext ctx, DatagramPacket msg, List<Object> out) throws Exception {
    //Create our response packet from the datagram we received
    final SourceResponsePacket packet = builder.construct(msg.content());
    if (packet != null) {
        try {
            SourceServerResponse response = SourceResponseFactory.createResponseFrom(packet);
            if (response != null) {
                response.setSender(msg.sender());
                response.setRecipient(msg.recipient());
                response.setResponsePacket(packet);
                log.debug("Receiving Data '{}' from '{}' using Channel Id: {}", response.getClass().getSimpleName(), ctx.channel().remoteAddress(), ctx.channel().id());
                //Pass the message back to the messenger
                responseHandler.accept(response, null);
                return;
            }
        } catch (Exception e) {
            responseHandler.accept(null, new AsyncGameLibCheckedException("Error while decoding source query response", e));
        }
    }
    throw new NoResponseFoundForPacket("Could not find a response handler for the received datagram packet", packet);
}

@Override
protected void decode(ChannelHandlerContext ctx, DatagramPacket packet, List<Object> list) throws Exception {
    // Requires a session
    McpeSession session = server.getSessionManager().get(packet.sender());

    if (session == null)
        return;

    packet.content().markReaderIndex();
    RakNetDatagramFlags flags = new RakNetDatagramFlags(packet.content().readByte());
    packet.content().resetReaderIndex();

    if (flags.isValid() && !flags.isAck() && !flags.isNak()) {
        RakNetDatagram datagram = new RakNetDatagram();
        datagram.decode(packet.content());
        list.add(new AddressedRakNetDatagram(datagram, packet.recipient(), packet.sender()));
    }
}

/**
 * 释放所有内存
 */
private void release()
{
  for (DatagramPacket dp : this.inputs)
  {
    dp.release();
  }
  this.inputs.clear();
  for (KcpOnUdp ku : this.kcps.values())
  {
    if (!ku.isClosed())
    {
      ku.release();
    }
  }
  this.kcps.clear();
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
    final SocketAddress remoteAddress = msg.sender();
    final CodecAggregator.Result result;
    try (Timer.Context ignored = aggregationTimer.time()) {
        result = aggregator.addChunk(msg.content(), remoteAddress);
    }
    final ByteBuf completeMessage = result.getMessage();
    if (completeMessage != null) {
        LOG.debug("Message aggregation completion, forwarding {}", completeMessage);
        ctx.fireChannelRead(completeMessage);
    } else if (result.isValid()) {
        LOG.debug("More chunks necessary to complete this message");
    } else {
        invalidChunksMeter.mark();
        LOG.debug("Message chunk was not valid and discarded.");
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
    String udpContentString = packet.content().toString(CharsetUtil.UTF_8);

    long currentTimestampInMilliseconds = System.currentTimeMillis();

    List<StatsdMetric> statsdMetrics = StatsdMetric.parseStatsdMetrics(udpContentString, currentTimestampInMilliseconds); 

    for (StatsdMetric statsdMetric : statsdMetrics) {
        long hashKey = GlobalVariables.metricHashKeyGenerator.incrementAndGet();
        statsdMetric.setHashKey(hashKey);

        if (statsdMetric.getMetricTypeCode() == StatsdMetric.GAUGE_TYPE) GlobalVariables.statsdGaugeMetrics.put(statsdMetric.getHashKey(), statsdMetric);
        else GlobalVariables.statsdNotGaugeMetrics.put(statsdMetric.getHashKey(), statsdMetric);

        if (statsdMetric.getBucket() != null) statsdMetric.getBucket().hashCode();

        GlobalVariables.incomingMetricsCount.incrementAndGet();
    }

    if (ApplicationConfiguration.isDebugModeEnabled()) {
        logger.info("UDP_Statsd_Received_Metrics=" + statsdMetrics.size());
        logger.info("UDP_Statsd_String=\"" + udpContentString + "\"");
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
    String udpContentString = packet.content().toString(CharsetUtil.UTF_8);

    long currentTimestampInMilliseconds = System.currentTimeMillis();

    List<GraphiteMetric> graphiteMetrics = GraphiteMetric.parseGraphiteMetrics(udpContentString, 
            GlobalVariables.graphitePassthroughPrefix, currentTimestampInMilliseconds);

    for (GraphiteMetric graphiteMetric : graphiteMetrics) {
        long hashKey = GlobalVariables.metricHashKeyGenerator.incrementAndGet();
        graphiteMetric.setHashKey(hashKey);
        GlobalVariables.graphitePassthroughMetrics.put(graphiteMetric.getHashKey(), graphiteMetric);
        GlobalVariables.incomingMetricsCount.incrementAndGet();
    }

    if (ApplicationConfiguration.isDebugModeEnabled()) {
        logger.info("UDP_Graphite_Passthrough_Received_Metrics=" + graphiteMetrics.size());
        logger.info("UDP_Graphite_Passthrough_String=\"" + udpContentString + "\"");
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
    String udpContentString = packet.content().toString(CharsetUtil.UTF_8);

    long currentTimestampInMilliseconds = System.currentTimeMillis();

    List<GraphiteMetric> graphiteMetrics = GraphiteMetric.parseGraphiteMetrics(udpContentString, 
            GlobalVariables.graphiteAggregatedPrefix, currentTimestampInMilliseconds);

    for (GraphiteMetric graphiteMetric : graphiteMetrics) {
        long hashKey = GlobalVariables.metricHashKeyGenerator.incrementAndGet();
        graphiteMetric.setHashKey(hashKey);
        if (graphiteMetric.getMetricPath() != null) graphiteMetric.getMetricPath().hashCode();
        GlobalVariables.graphiteAggregatorMetrics.put(graphiteMetric.getHashKey(), graphiteMetric);
        GlobalVariables.incomingMetricsCount.incrementAndGet();
    }

    if (ApplicationConfiguration.isDebugModeEnabled()) {
        logger.info("UDP_Graphite_Aggregator_Received_Metrics=" + graphiteMetrics.size());
        logger.info("UDP_Graphite_Aggregator_String=\"" + udpContentString + "\"");
    }
}

private void handleBasicStats(ChannelHandlerContext ctx, DatagramPacket packet, int sessionId) {
    LanternServer server = this.queryServer.getGame().getServer();

    // TODO: Find out how to support the size and max size properties
    final Cause cause = Cause.of(EventContext.empty(),
            new SimpleRemoteConnection((InetSocketAddress) ctx.channel().remoteAddress(), null));
    final QueryServerEvent.Basic event = SpongeEventFactory.createQueryServerEventBasic(cause,
            (InetSocketAddress) ctx.channel().localAddress(), "SMP", this.getWorldName(), server.getMotd().toPlain(),
            server.getMaxPlayers(), Integer.MAX_VALUE, server.getOnlinePlayers().size(), 0);
    Sponge.getEventManager().post(event);

    final InetSocketAddress address = event.getAddress();

    final ByteBuf buf = ctx.alloc().buffer();
    buf.writeByte(ACTION_STATS);
    buf.writeInt(sessionId);
    writeString(buf, event.getMotd());
    writeString(buf, event.getGameType());
    writeString(buf, event.getMap());
    writeString(buf, String.valueOf(event.getPlayerCount()));
    writeString(buf, String.valueOf(event.getMaxPlayerCount()));
    buf.writeShortLE(address.getPort());
    writeString(buf, address.getHostString());
    ctx.write(new DatagramPacket(buf, packet.sender()));
}

@Override
protected void encode(ChannelHandlerContext ctx, SipResponse msg, List<Object> out)
{
    ByteBuf buff = ctx.alloc().buffer();
    buff.writeBytes(msg.getVersion().toString().getBytes(CharsetUtil.UTF_8));
    buff.writeBytes(" ".getBytes(CharsetUtil.UTF_8));
    buff.writeBytes(String.valueOf(msg.getStatusCode()).getBytes(CharsetUtil.UTF_8));
    buff.writeBytes(" ".getBytes(CharsetUtil.UTF_8));
    buff.writeBytes(msg.getMessage().getBytes(CharsetUtil.UTF_8));
    buff.writeBytes(" ".getBytes(CharsetUtil.UTF_8));
    buff.writeBytes("\n".getBytes(CharsetUtil.UTF_8));
    msg.getHeaders().forEach((k, v) ->
    {
        v.stream().forEach(val ->
        {
            buff.writeBytes(k.getBytes(CharsetUtil.UTF_8));
            buff.writeBytes(": ".getBytes(CharsetUtil.UTF_8));
            buff.writeBytes(val.getBytes(CharsetUtil.UTF_8));
            buff.writeBytes("\n".getBytes(CharsetUtil.UTF_8));
        });
    });
    buff.writeBytes("\n".getBytes(CharsetUtil.UTF_8));
    DatagramPacket packet = new DatagramPacket(buff, msg.getRecipient());
    out.add(packet);
}

public void run(int port) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
    Bootstrap b = new Bootstrap();
    b.group(group).channel(NioDatagramChannel.class)
        .option(ChannelOption.SO_BROADCAST, true)
        .handler(new ChineseProverbClientHandler());
    Channel ch = b.bind(0).sync().channel();
    // 向网段内的所有机器广播UDP消息
    ch.writeAndFlush(
        new DatagramPacket(Unpooled.copiedBuffer("谚语字典查询?",
            CharsetUtil.UTF_8), new InetSocketAddress(
            "255.255.255.255", port))).sync();
    if (!ch.closeFuture().await(15000)) {
    System.out.println("查询超时!");
    }
} finally {
    group.shutdownGracefully();
}
   }

@Override
protected Object filterOutboundMessage(Object msg) {
    if (msg instanceof DatagramPacket || msg instanceof ByteBuf) {
        return msg;
    }

    if (msg instanceof AddressedEnvelope) {
        @SuppressWarnings("unchecked")
        AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
        if (e.content() instanceof ByteBuf) {
            return msg;
        }
    }

    throw new UnsupportedOperationException(
            "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
}

/**
 * Try to add the given {@link DatagramPacket}. Returns {@code true} on success,
 * {@code false} otherwise.
 */
boolean add(DatagramPacket packet) {
    if (count == packets.length) {
        return false;
    }
    ByteBuf content = packet.content();
    int len = content.readableBytes();
    if (len == 0) {
        return true;
    }
    NativeDatagramPacket p = packets[count];
    InetSocketAddress recipient = packet.recipient();
    if (!p.init(content, recipient)) {
        return false;
    }

    count++;
    return true;
}

@Override
public boolean send(Message message) {
    if (message.payload == null) {
        log.error("Payload of message {} may not be null.", message.toString());
        return false;
    }
    if (message.socketAddressRecipient == null) {
        log.error("Recipient of message {} may not be null.", message.toString());
        return false;
    }
    if (config.trackData) {
        int frame = 0;
        config.netStats.getData().add(
                new NetStats.Line(true, message.getSenderId(), frame, message.getTimestamp(), message.getClass(), ((ByteBuf)message.payload).writerIndex()));
    }

    channel.writeAndFlush(new DatagramPacket((ByteBuf) message.payload, message.socketAddressRecipient))
            .addListener(new FutureGenericFutureListener("writeAndFlush", KryoNettyPeer.this, message));
    return true;
}

@Override
public String resolve(DatagramPacket packet, ChannelHandlerContext ctx) throws KeyNotResolvedException {
    try {
        packet.content().retain();
        int readableBytes = packet.content().readableBytes();
        byte[] bytes = new byte[readableBytes];
        packet.content().readBytes(bytes);
        packet.content().resetReaderIndex();
        StunMessage stunMessage = StunMessage.decode(bytes, (char) 0, (char) bytes.length);
        UsernameAttribute attribute = (UsernameAttribute)stunMessage.getAttribute(StunAttribute.USERNAME);

        return new String(attribute.getUsername(), "UTF-8");
    } catch (StunException | UnsupportedEncodingException e) {
        throw new KeyNotResolvedException();
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg)  {
    try {
        String configuration = msg.content().toString(CharsetUtil.UTF_8);
        RemoteConfiguration remoteConfiguration = RemoteConfiguration.buildRemoteConfiguration(configuration);

        remoteControlCache.put("rtp:"+remoteConfiguration.getSsrc(), remoteConfiguration.getRtpPort());
        remoteControlCache.put("rtp:"+remoteConfiguration.getStunUsername(), remoteConfiguration.getRtpPort());
        remoteControlCache.put("rtcp:"+remoteConfiguration.getSsrc(), remoteConfiguration.getRtcpPort());
        remoteControlCache.put("rtcp:"+remoteConfiguration.getStunUsername(), remoteConfiguration.getRtcpPort());

        dtlsCache.put(remoteConfiguration.getRtpPort(), remoteConfiguration.getSsrc());
        dtlsCache.put(remoteConfiguration.getRtcpPort(), remoteConfiguration.getSsrc());

        ctx.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("SUCCESS", CharsetUtil.UTF_8), msg.sender()));
    } catch (Exception e) {
        ctx.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("ERROR", CharsetUtil.UTF_8), msg.sender()));
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
    try {
        msg.content().retain();
        int readableBytes = msg.content().readableBytes();
        byte[] bytes = new byte[readableBytes];
        msg.content().readBytes(bytes);

        StunMessage stunMessage = StunMessage.decode(bytes, (char) 0, (char) bytes.length);

        System.out.println("Stun message type - "  + stunMessage.getClass());
    } catch (StunException e) {
        e.printStackTrace();
    }

    ctx.close();
}

public static void main(String[] args) {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new RtpPartyAHandler());

        ch = b.bind(36001).sync().channel();

        ch.writeAndFlush(new DatagramPacket(
                Unpooled.copiedBuffer(rtpDataByeArray),
                new InetSocketAddress("127.0.0.1", PORT))).sync();

        if (!ch.closeFuture().await(5000)) {
            Assert.fail("Rtp communication timeout");
        } else {

        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        group.shutdownGracefully();
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    DatagramPacket packet = (DatagramPacket) msg;
    try {

        InputStream stream = new ByteBufInputStream(packet.content());
        Object data = new CompactObjectInputStream(stream, ClassResolvers.cacheDisabled(null)).readObject();
        MDC.put("id", LogSourceId.getInstance().getId());
        logger.callAppenders(((LoggingEventWrapper) data).event);

    } catch (Throwable e){
        System.out.println(e);
    }
    ReferenceCountUtil.release(msg);
}

@Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if(msg instanceof DatagramPacket){
            ctx.write(msg, promise);
            return;
        }

        ByteBuf buf = ctx.alloc().heapBuffer();
//      int startIdx = buf.writerIndex();

        ByteBufOutputStream bout = new ByteBufOutputStream(buf);
//      bout.write(LENGTH_PLACEHOLDER);
        ObjectOutputStream oout = new CompactObjectOutputStream(bout);
        oout.writeObject(msg);
        oout.flush();
        oout.close();
//      int endIdx = buf.writerIndex();

//      buf.setInt(startIdx, endIdx - startIdx - 4);
        Object data = new DatagramPacket(buf, new InetSocketAddress("255.255.255.255", port));
        ctx.write(data, promise);
    }

/**
   * Decodes a {@link DatagramPacket} to a {@link DataPacket} wrapped into an {@link AddressedEnvelope} to allow multicast on
   * the used {@link SocketChannel}. 
   * 
   * @param ctx The context of the ChannelHandler
   * @param msg the message which should be encoded
   * @param out a list where all messages are written to
   */
  @Override
  protected void decode(ChannelHandlerContext ctx, DatagramPacket msg, List<Object> out) throws Exception {
    final ByteBuf content = msg.content();
final SocketAddress sender = msg.sender();
final SocketAddress recipient = msg.recipient();

      try {
          final DataPacket dataPacket = DataPacket.decode(content);
          final AddressedEnvelope<DataPacket, SocketAddress> newMsg = 
                new DefaultAddressedEnvelope<>(
                        dataPacket, recipient, sender);
        out.add(newMsg);
      } catch (Exception e) {
          LOG.debug("Failed to decode RTP packet.", e);
      }
  }

public void run(int port) throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new ChineseProverbClientHandler());
        Channel ch = b.bind(0).sync().channel();
        // 向网段内的所有机器广播UDP消息
        ch.writeAndFlush(
                new DatagramPacket(Unpooled.copiedBuffer("谚语字典查询?",
                        CharsetUtil.UTF_8), new InetSocketAddress(
                        "255.255.255.255", port))).sync();
        if (!ch.closeFuture().await(15000)) {
            System.out.println("查询超时!");
        }
    } finally {
        group.shutdownGracefully();
    }
}

@Override
public void process(DatagramPacket packet) throws Exception {
  long total = totalPackets.incrementAndGet();
  boolean droppedPacket = false;
  ParseResult result;
  try {
    List<Record> records = parser.parse(packet.content(), packet.recipient(), packet.sender());
    result = new ParseResult(records);
  } catch (OnRecordErrorException ex) {
    result = new ParseResult(ex);
  }
  if (!queue.offer(result)) {
    droppedPacket = true;
    long dropped = droppedPackets.incrementAndGet();
    if (dropped % 1000 == 0) {
      LOG.info("Could not add packet to queue, dropped {} of {} packets", dropped, total);
    }
  }
  if (!droppedPacket && total % 1000 == 0) {
    LOG.info("Consumed {} total packets", total);
  }
}

public byte[] serialize(UDPMessage message) throws IOException {
  baos.reset();
  ObjectOutputStream oos = new ObjectOutputStream(baos);
  oos.writeInt(UDPConstants.UDP_MESSAGE_VERSION);
  oos.writeInt(message.getType());
  oos.writeLong(message.getReceived());
  DatagramPacket datagram = message.getDatagram();
  oos.writeUTF(datagram.sender().getAddress().getHostAddress());
  oos.writeInt(datagram.sender().getPort());
  oos.writeUTF(datagram.recipient().getAddress().getHostAddress());
  oos.writeInt(datagram.recipient().getPort());
  if (datagram.content().readableBytes() > MAX_UDP_PACKAGE_SIZE) {
    throw new IOException(Utils.format("Message size '{}' exceeds maximum size '{}'",
        baos.size(),
        MAX_UDP_PACKAGE_SIZE
    ));
  }
  oos.writeInt(datagram.content().readableBytes());
  datagram.content().readBytes(oos, datagram.content().readableBytes());
  oos.close();
  byte[] buffer = new byte[baos.size()];
  System.arraycopy(baos.getInternalBuffer(), 0, buffer, 0, baos.size());
  return buffer;
}

public UDPMessage deserialize(byte[] buffer) throws IOException {
  ByteArrayInputStream bais = new ByteArrayInputStream(buffer);
  ObjectInputStream ois = new ObjectInputStream(bais);
  int version = ois.readInt();
  if (version == UDPConstants.UDP_MESSAGE_VERSION) {
    int type = ois.readInt();
    long received = ois.readLong();
    String address = ois.readUTF();
    int port = ois.readInt();
    InetSocketAddress sender = new InetSocketAddress(address, port);
    address = ois.readUTF();
    port = ois.readInt();
    InetSocketAddress receiver = new InetSocketAddress(address, port);
    int dataLen = ois.readInt();
    byte[] data = new byte[dataLen];
    ois.readFully(data);
    ois.close();
    ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
    DatagramPacket datagram = new DatagramPacket(byteBuf, receiver, sender);
    return new UDPMessage(type, received, datagram);
  } else {
    throw new IOException(Utils.format("Unsupported version '{}'", version));
  }
}

@Override
public void process(DatagramPacket packet) throws Exception {
  LOG.debug(
      "Datagram from '{}:{}' accepted",
      packet.sender().getAddress().getHostAddress(),
      packet.sender().getPort()
  );
  if (!isQueueOverLimit()) {
    getExecutorService().submit(createDispacher(packet));
    acceptedPackagesMeter.mark();
  } else {
    discardedPackagesMeter.mark();
    String msg = Utils.format("Datagram from '{}:{}' discarded, queue over '{}'",
        packet.sender().getAddress().getHostAddress(),
        packet.sender().getPort(),
        getQueueLimit()
    );
    LOG.warn(msg);
    getErrorQueue().offer(new Exception(msg));
  }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof DatagramPacket) {
        // Get packet and sender data
        DatagramPacket datagram = (DatagramPacket) msg;
        InetSocketAddress sender = datagram.sender();
        RakNetPacket packet = new RakNetPacket(datagram);

        // If an exception happens it's because of this address
        this.causeAddress = sender;

        // Is the sender blocked?
        if (this.addressBlocked(sender.getAddress())) {
            BlockedAddress status = blocked.get(sender.getAddress());
            if (status.getTime() <= BlockedAddress.PERMANENT_BLOCK) {
                return; // Permanently blocked
            }
            if (System.currentTimeMillis() - status.getStartTime() < status.getTime()) {
                return; // Time hasn't expired
            }
            this.unblockAddress(sender.getAddress());
        }

        // Handle the packet and release the buffer
        server.handleMessage(packet, sender);
        datagram.content().readerIndex(0); // Reset position
        server.getListener().handleNettyMessage(datagram.content(), sender);
        datagram.content().release(); // No longer needed

        // No exceptions occurred, release the suspect
        this.causeAddress = null;
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    Channel udpChannel = XChannelMapper.getUdpChannel(udpSource);
    if (udpChannel == null) {
        log.warn("Bad Connection! (udp channel closed)");
        XChannelMapper.closeChannelGracefullyByTcpChannel(ctx.channel());
    } else if (udpChannel.isActive()) {
        ByteBuf byteBuf = (ByteBuf) msg;
        try {
            if (!byteBuf.hasArray()) {
                byte[] bytes = new byte[byteBuf.readableBytes()];
                byteBuf.getBytes(0, bytes);
                bytes = wrapper.unwrap(bytes);
                XRequest request = requestResolver.parse(bytes);
                String host = request.getHost();
                int port = request.getPort();
                byte[] content = Arrays.copyOfRange(bytes, bytes.length - request.getSubsequentDataLength(), bytes.length);
                log.info("\t          Proxy << Target \tFrom   {}:{}", host, port);

                // redirect tcp -> udp
                udpChannel.writeAndFlush(new DatagramPacket(Unpooled.wrappedBuffer(content), udpSource, new InetSocketAddress(host, port)));
                log.info("\tClient << Proxy           \tGet [{} bytes]", content.length);
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
}