Java 类io.netty.buffer.PooledByteBufAllocator 实例源码

/**
 * Create a pooled ByteBuf allocator but disables the thread-local cache. Thread-local caches
 * are disabled for TransportClients because the ByteBufs are allocated by the event loop thread,
 * but released by the executor thread rather than the event loop thread. Those thread-local
 * caches actually delay the recycling of buffers, leading to larger memory usage.
 */
public static PooledByteBufAllocator createPooledByteBufAllocator(
    boolean allowDirectBufs,
    boolean allowCache,
    int numCores) {
  if (numCores == 0) {
    numCores = Runtime.getRuntime().availableProcessors();
  }
  return new PooledByteBufAllocator(
      allowDirectBufs && PlatformDependent.directBufferPreferred(),
      Math.min(getPrivateStaticField("DEFAULT_NUM_HEAP_ARENA"), numCores),
      Math.min(getPrivateStaticField("DEFAULT_NUM_DIRECT_ARENA"), allowDirectBufs ? numCores : 0),
      getPrivateStaticField("DEFAULT_PAGE_SIZE"),
      getPrivateStaticField("DEFAULT_MAX_ORDER"),
      allowCache ? getPrivateStaticField("DEFAULT_TINY_CACHE_SIZE") : 0,
      allowCache ? getPrivateStaticField("DEFAULT_SMALL_CACHE_SIZE") : 0,
      allowCache ? getPrivateStaticField("DEFAULT_NORMAL_CACHE_SIZE") : 0
  );
}

/**
 * <b>一定要 {@link ByteBuf#release()}
 */
public static ByteBuf encode(RemotingCommand cmd) {
    int reqId = cmd.getOpaque();
    byte[] body = cmd.getBody();
    HashMap<String, String> msgs = cmd.getExtFields();
    byte[] append = JsonUtil.toBytes(msgs);
    int initialCapacity = 4 + 4 // total size+reqId
            + 4 + body.length // body
            + 4 + append.length;// apend
    ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(initialCapacity);
    buf.writeInt(initialCapacity);
    buf.writeInt(reqId);
    buf.writeInt(body.length);
    buf.writeBytes(body);
    buf.writeInt(append.length);
    buf.writeBytes(append);

    return buf;
}

protected AbstractNettyServer(String serverName) {
    this.serverName = Objects.requireNonNull(serverName, "server name");
    bootstrap = new ServerBootstrap();
    if (Epoll.isAvailable()) {
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024).channel(EpollServerSocketChannel.class)
                .childOption(ChannelOption.SO_LINGER, 0).childOption(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true);
        log.info(serverName + " epoll init");
    } else {
        bootstrap.channel(NioServerSocketChannel.class);
        log.info(serverName + " nio init");
    }
    bootstrap.group(bossGroup, workerGroup).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, true).childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    initPipeline(ch.pipeline());
                }
            });
}

private BinaryClient(BinaryClientBuilder builder) throws Exception {
    this.clientName = builder.clientName;
    this.remoteAddress = Objects.requireNonNull(builder.remoteAddress, "remoteAddress");
    this.autoReconnect = builder.autoReconnect;
    this.decoder = Objects.requireNonNull(builder.decoder, "decoder");
    this.encoder = Objects.requireNonNull(builder.encoder, "encoder");
    this.factory = Objects.requireNonNull(builder.factory, "factory");
    this.onChannelStateChanged = builder.onChannelStateChanged;
    this.onExceptionCaught = builder.onExceptionCaught;
    this.onConnectionEffective = builder.onConnectionEffective;
    this.dispatchMessage = builder.dispatchMessage;
    this.heartIntervalSec = builder.heartIntervalSec;
    // 内部消息注册
    factory.registerMsg(new ConnectionValidateServerHandler())
            .registerMsg(new ConnectionValidateSuccessServerHandler()).registerMsg(new HeartServerHandler());
    decodeUtil = SymmetricEncryptionUtil.getDecodeInstance(remoteAddress.getPass());
    bootstrap = new Bootstrap();
    bootstrap.channel(NioSocketChannel.class);
    log.info(clientName + " nio init");
    bootstrap.group(group).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .handler(new ChannelInitializerImpl());
}

/**
 * 启动服务
 *
 * @throws Exception 异常
 */
public void start() throws Exception {
    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(channelInitializer)
                .option(ChannelOption.SO_BACKLOG, aceServerConfig.getBackSize())
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        ChannelFuture future = bootstrap.bind(aceServerConfig.getPort()).sync();
        System.out.println("ace server starter on port : " + aceServerConfig.getPort());
        future.channel().closeFuture().sync();
    } finally {
        close();
    }


}

/**
 * Create a pooled ByteBuf allocator but disables the thread-local cache. Thread-local caches
 * are disabled for TransportClients because the ByteBufs are allocated by the event loop thread,
 * but released by the executor thread rather than the event loop thread. Those thread-local
 * caches actually delay the recycling of buffers, leading to larger memory usage.
 */
public static PooledByteBufAllocator createPooledByteBufAllocator(
    boolean allowDirectBufs,
    boolean allowCache,
    int numCores) {
  if (numCores == 0) {
    numCores = Runtime.getRuntime().availableProcessors();
  }
  return new PooledByteBufAllocator(
    allowDirectBufs && PlatformDependent.directBufferPreferred(),
    Math.min(getPrivateStaticField("DEFAULT_NUM_HEAP_ARENA"), numCores),
    Math.min(getPrivateStaticField("DEFAULT_NUM_DIRECT_ARENA"), allowDirectBufs ? numCores : 0),
    getPrivateStaticField("DEFAULT_PAGE_SIZE"),
    getPrivateStaticField("DEFAULT_MAX_ORDER"),
    allowCache ? getPrivateStaticField("DEFAULT_TINY_CACHE_SIZE") : 0,
    allowCache ? getPrivateStaticField("DEFAULT_SMALL_CACHE_SIZE") : 0,
    allowCache ? getPrivateStaticField("DEFAULT_NORMAL_CACHE_SIZE") : 0
  );
}

private void startAcceptingConnections() throws InterruptedException {
    ServerBootstrap b = new ServerBootstrap();
    b.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
    b.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
    b.option(ChannelOption.SO_RCVBUF, 1048576);
    b.option(ChannelOption.TCP_NODELAY, true);
    b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    b.group(serverGroup, clientGroup);
    b.channel(serverChannelClass);
    if (enableNettyTls) {
        b.childHandler(new SslServerCommunicationChannelInitializer());
    } else {
        b.childHandler(new OnosCommunicationChannelInitializer());
    }
    b.option(ChannelOption.SO_BACKLOG, 128);
    b.childOption(ChannelOption.SO_KEEPALIVE, true);

    // Bind and start to accept incoming connections.
    b.bind(localEp.port()).sync().addListener(future -> {
        if (future.isSuccess()) {
            log.info("{} accepting incoming connections on port {}", localEp.host(), localEp.port());
        } else {
            log.warn("{} failed to bind to port {}", localEp.host(), localEp.port(), future.cause());
        }
    });
}

@Override
        protected void initChannel(SocketChannel ch) throws Exception {
            try {
                ch.config().setOption(ChannelOption.IP_TOS, 0x18);
//                ch.config().setOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
//                ch.config().setOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
            } catch (ChannelException ex) {
                // IP_TOS not supported by platform, ignore
            }
            ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

            PacketRegistry r = new PacketRegistry();

            ch.pipeline().addLast(new HttpServerCodec());
            ch.pipeline().addLast(new HttpObjectAggregator(65536));
            ch.pipeline().addLast(new WebSocketHandler());
            ch.pipeline().addLast(new PacketDecoder(r));
            ch.pipeline().addLast(new PacketEncoder(r));
            ch.pipeline().addLast(new ClientHandler(server));
        }

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.TCP_NODELAY, true);
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    PacketRegistry r = new PacketRegistry();

    ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(0, 0, 30));
    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder(r));
    ch.pipeline().addLast(new PacketEncoder(r));
    ch.pipeline().addLast(mExecutorGroup, "serverHandler", new ClientHandler(mServer));
}

private void createClient(Listener listener, EventLoopGroup workerGroup, ChannelFactory<? extends Channel> channelFactory) {
    this.workerGroup = workerGroup;
    this.bootstrap = new Bootstrap();
    bootstrap.group(workerGroup)//
            .option(ChannelOption.SO_REUSEADDR, true)//
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//
            .channelFactory(channelFactory);
    bootstrap.handler(new ChannelInitializer<Channel>() { // (4)
        @Override
        public void initChannel(Channel ch) throws Exception {
            initPipeline(ch.pipeline());
        }
    });
    initOptions(bootstrap);
    listener.onSuccess();
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder());
    ch.pipeline().addLast(new PacketEncoder());
    ch.pipeline().addLast(new ClientHandler(server));
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new Handshaker());
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder());
    ch.pipeline().addLast(new PacketEncoder());
    ch.pipeline().addLast(new ClientHandler(server));
}

private Bootstrap newBootstrap() {
    Bootstrap boot = new Bootstrap();
    boot.channel(NettyPlatformIndependent.channelClass());
    boot.option(ChannelOption.TCP_NODELAY, true);
    // replace by heart beat
    boot.option(ChannelOption.SO_KEEPALIVE, false);
    // default is pooled direct
    // ByteBuf(io.netty.util.internal.PlatformDependent.DIRECT_BUFFER_PREFERRED)
    boot.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    // 32kb(for massive long connections, See
    // http://www.infoq.com/cn/articles/netty-million-level-push-service-design-points)
    // 64kb(RocketMq remoting default value)
    boot.option(ChannelOption.SO_SNDBUF, 32 * 1024);
    boot.option(ChannelOption.SO_RCVBUF, 32 * 1024);
    // temporary settings, need more tests
    boot.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(8 * 1024, 32 * 1024));
    //default is true, reduce thread context switching
    boot.option(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP, true);
    return boot;
}

private ServerBootstrap newServerBootstrap() {
    ServerBootstrap serverBoot = new ServerBootstrap();
    serverBoot.channel(NettyPlatformIndependent.serverChannelClass());
    // connections wait for accept
    serverBoot.option(ChannelOption.SO_BACKLOG, 1024);
    serverBoot.option(ChannelOption.SO_REUSEADDR, true);
    // replace by heart beat
    serverBoot.childOption(ChannelOption.SO_KEEPALIVE, false);
    serverBoot.childOption(ChannelOption.TCP_NODELAY, true);
    serverBoot.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
    serverBoot.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
    // temporary settings, need more tests
    serverBoot.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(8 * 1024, 32 * 1024));
    serverBoot.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    //default is true, reduce thread context switching
    serverBoot.childOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP, true);
    return serverBoot;
}

@Override
public void initChannel(Channel ch) throws Exception
{
    try
    {
        ch.config().setOption( ChannelOption.IP_TOS, 0x18 );
    } catch ( ChannelException ex )
    {
        // IP_TOS is not supported (Windows XP / Windows Server 2003)
    }
    ch.config().setOption( ChannelOption.TCP_NODELAY, true );
    ch.config().setAllocator( PooledByteBufAllocator.DEFAULT );

    ch.pipeline().addLast( TIMEOUT_HANDLER, new ReadTimeoutHandler( BungeeCord.getInstance().config.getTimeout(), TimeUnit.MILLISECONDS ) );
    ch.pipeline().addLast( FRAME_DECODER, new Varint21FrameDecoder() );
    ch.pipeline().addLast( FRAME_PREPENDER, framePrepender );

    ch.pipeline().addLast( BOSS_HANDLER, new HandlerBoss() );
}

/**
 * starts server to handle discovery-request from client-channel
 *
 * @throws Exception
 */
public void startServer() throws Exception {

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    bootstrap.group(acceptorGroup, workerGroup);
    bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
    bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
            new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
    bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
    EventLoopUtil.enableTriggeredMode(bootstrap);

    bootstrap.childHandler(new ServiceChannelInitializer(this, config, false));
    // Bind and start to accept incoming connections.
    bootstrap.bind(config.getServicePort()).sync();
    LOG.info("Started Pulsar Discovery service on port {}", config.getServicePort());

    if (config.isTlsEnabled()) {
        ServerBootstrap tlsBootstrap = bootstrap.clone();
        tlsBootstrap.childHandler(new ServiceChannelInitializer(this, config, true));
        tlsBootstrap.bind(config.getServicePortTls()).sync();
        LOG.info("Started Pulsar Discovery TLS service on port {}", config.getServicePortTls());
    }
}

protected void connectWithMaster() {

        if (!(getLifecycleState().isStarting() || getLifecycleState().isStarted())) {
            logger.info("[connectWithMaster][do not connect, is stopped!!]{}", redisMaster.masterEndPoint());
            return;
        }

        Bootstrap b = new Bootstrap();
        b.group(slaveEventLoopGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new LoggingHandler(LogLevel.DEBUG));
                        p.addLast(new NettySimpleMessageHandler());
                        p.addLast(createHandler());
                    }
                });

        doConnect(b);
    }

public static AllocatorStats generate(String allocatorName) {
    PooledByteBufAllocator allocator = null;
    if ("default".equals(allocatorName)) {
        allocator = PooledByteBufAllocator.DEFAULT;
    } else if ("ml-cache".equals(allocatorName)) {
        allocator = EntryCacheImpl.allocator;
    } else {
        throw new IllegalArgumentException("Invalid allocator name : " + allocatorName);
    }

    AllocatorStats stats = new AllocatorStats();
    stats.directArenas = allocator.directArenas().stream().map(x -> newPoolArenaStats(x))
            .collect(Collectors.toList());
    stats.heapArenas = allocator.heapArenas().stream().map(x -> newPoolArenaStats(x)).collect(Collectors.toList());

    stats.numDirectArenas = allocator.numDirectArenas();
    stats.numHeapArenas = allocator.numHeapArenas();
    stats.numThreadLocalCaches = allocator.numThreadLocalCaches();
    stats.normalCacheSize = allocator.normalCacheSize();
    stats.smallCacheSize = allocator.smallCacheSize();
    stats.tinyCacheSize = allocator.tinyCacheSize();
    return stats;
}

public static byte[] createMessageWrittenToLedger(String msg) throws Exception {
    PulsarApi.MessageMetadata.Builder messageMetadataBuilder = PulsarApi.MessageMetadata.newBuilder();
    messageMetadataBuilder.setPublishTime(System.currentTimeMillis());
    messageMetadataBuilder.setProducerName("createMessageWrittenToLedger");
    messageMetadataBuilder.setSequenceId(1);
    PulsarApi.MessageMetadata messageMetadata = messageMetadataBuilder.build();
    ByteBuf data = UnpooledByteBufAllocator.DEFAULT.heapBuffer().writeBytes(msg.getBytes());

    int msgMetadataSize = messageMetadata.getSerializedSize();
    int payloadSize = data.readableBytes();
    int totalSize = 4 + msgMetadataSize + payloadSize;

    ByteBuf headers = PooledByteBufAllocator.DEFAULT.heapBuffer(totalSize, totalSize);
    ByteBufCodedOutputStream outStream = ByteBufCodedOutputStream.get(headers);
    headers.writeInt(msgMetadataSize);
    messageMetadata.writeTo(outStream);
    ByteBuf headersAndPayload = DoubleByteBuf.get(headers, data);
    byte[] byteMessage = headersAndPayload.nioBuffer().array();
    headersAndPayload.release();
    return byteMessage;
}

/**
 * Verify that readableBytes() returns writerIndex - readerIndex. In this case writerIndex is the end of the buffer
 * and readerIndex is increased by 64.
 *
 * @throws Exception
 */
@Test
public void testReadableBytes() throws Exception {

    ByteBuf b1 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
    b1.writerIndex(b1.capacity());
    ByteBuf b2 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
    b2.writerIndex(b2.capacity());
    ByteBuf buf = DoubleByteBuf.get(b1, b2);

    assertEquals(buf.readerIndex(), 0);
    assertEquals(buf.writerIndex(), 256);
    assertEquals(buf.readableBytes(), 256);

    for (int i = 0; i < 4; ++i) {
        buf.skipBytes(64);
        assertEquals(buf.readableBytes(), 256 - 64 * (i + 1));
    }

    buf.release();

    assertEquals(buf.refCnt(), 0);
    assertEquals(b1.refCnt(), 0);
    assertEquals(b2.refCnt(), 0);
}

@Test
public void testCapacity() throws Exception {

    ByteBuf b1 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
    b1.writerIndex(b1.capacity());
    ByteBuf b2 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
    b2.writerIndex(b2.capacity());
    ByteBuf buf = DoubleByteBuf.get(b1, b2);

    assertEquals(buf.capacity(), 256);
    assertEquals(buf.maxCapacity(), 256);

    buf.release();

    assertEquals(buf.refCnt(), 0);
    assertEquals(b1.refCnt(), 0);
    assertEquals(b2.refCnt(), 0);
}

@Test(dataProvider = "codec")
void testMultpileUsages(CompressionType type) throws IOException {
    CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(type);
    byte[] data = text.getBytes();

    for (int i = 0; i < 5; i++) {
        ByteBuf raw = PooledByteBufAllocator.DEFAULT.buffer();
        raw.writeBytes(data);
        ByteBuf compressed = codec.encode(raw);
        assertEquals(raw.readableBytes(), data.length);

        int compressedSize = compressed.readableBytes();

        ByteBuf uncompressed = codec.decode(compressed, data.length);

        assertEquals(compressed.readableBytes(), compressedSize);

        assertEquals(uncompressed.readableBytes(), data.length);
        assertEquals(uncompressed, raw);

        raw.release();
        compressed.release();
        uncompressed.release();
    }
}

@Test
    public void testNettyInternalBuffer() throws IOException{

        ByteBufAllocator allocator = new PooledByteBufAllocator(true);

        final ByteBuf byteBuf = allocator.buffer(1 << 10);
        byteBuf.writeBytes("1234567890".getBytes());

        System.out.println(byteBuf.readableBytes());

        scheduled.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
//              ByteBuffer byteBuffer = byteBuf.internalNioBuffer(0, byteBuf.readableBytes());
                byteBuf.nioBuffers();
            }
        }, 0, 100, TimeUnit.MICROSECONDS);

        System.out.println(byteBuf.readableBytes());

        waitForAnyKeyToExit();
    }

@SuppressWarnings("unchecked")
@Test
public void onErrorWhenGettingNodeOne() {
    HttpClientResponse<ByteBuf> urlsResponse = mock(HttpClientResponse.class);
    ByteBuf byteBuf = (new PooledByteBufAllocator()).directBuffer();
    ByteBufUtil.writeUtf8(byteBuf, onePactSource);
    when(urlsResponse.getContent()).thenReturn(Observable.just(byteBuf));
    when(urlsResponse.getStatus()).thenReturn(HttpResponseStatus.OK);

    when(rxClient.submit(any(RxClient.ServerInfo.class), any(HttpClientRequest.class)))
            .thenReturn(Observable.just(urlsResponse), Observable.error(new RuntimeException()));

    TestSubscriber<Node> testSubscriber = new TestSubscriber<>();
    pactsAggregator.aggregateNodes().toBlocking().subscribe(testSubscriber);
    testSubscriber.assertError(RuntimeException.class);

    verify(publisher).publishEvent(any(SystemEvent.class));
}

public NioClient(final URI uri) {
    super("nio-client-%d");
    final Bootstrap b = new Bootstrap().group(group);
    b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

    try {
        final MessageSerializer serializer = new GryoMessageSerializerV1d0();
        b.channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(final SocketChannel ch) {
                        final ChannelPipeline p = ch.pipeline();
                        p.addLast(
                                new NioGremlinResponseDecoder(serializer),
                                new NioGremlinRequestEncoder(true, serializer),
                                callbackResponseHandler);
                    }
                });

        channel = b.connect(uri.getHost(), uri.getPort()).sync().channel();
    } catch (Exception ex) {
        throw new RuntimeException(ex);
    }
}

public void start() {
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator(true))
                .childOption(ChannelOption.SO_REUSEADDR, true)
                .channel(NioServerSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(initializer);
        b.bind(port).sync();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Override
public byte[] getPayload() {
    String filterString = this.filter.toString();
    int payloadSize = (3 + filterString.length() + (this.startIp.length()));
    final ByteBuf payload = PooledByteBufAllocator.DEFAULT.buffer(payloadSize);
    try {
        payload.writeByte(getRegion());
        payload.writeBytes(getStartIp().getBytes());
        payload.writeByte(0); //terminating byte
        payload.writeBytes(filterString.getBytes());
        byte[] payloadBytes = new byte[payload.readableBytes()];
        payload.readBytes(payloadBytes);
        return payloadBytes;
    } finally {
        payload.release();
    }
}

public NomadServer(NomadLobby nLobby, EventLoopGroup bossGroup, EventLoopGroup workerGroup,
        EventExecutorGroup executorGroup) {
    sb = new ServerBootstrap();
    sb.group(bossGroup, workerGroup);
    sb.channel(NioServerSocketChannel.class);

    final int BUF_PER_CLIENT = Packet.MAX_PACKET_LENGTH * 4;
    final int MAX_CLIENTS = 2000;

    sb.option(ChannelOption.SO_BACKLOG, MAX_CLIENTS);
    sb.option(ChannelOption.SO_REUSEADDR, true);
    sb.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    sb.childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(BUF_PER_CLIENT));
    sb.childOption(ChannelOption.SO_SNDBUF, 65535);
    sb.childOption(ChannelOption.SO_RCVBUF, 65535);

    sb.childHandler(new ServerHandler(nLobby, executorGroup));
    String ip = Nomad.BIND_ON_ALL ? "0.0.0.0" : nLobby.getLobby().getIp();
    sb.localAddress(ip, nLobby.getLobby().getPort());
}

public byte[] decrypt(byte[] bytes) {
    ByteBuf c = null, p = null;
    byte[] result = null;
    try {
        c = Unpooled.wrappedBuffer(bytes);
        p = PooledByteBufAllocator.DEFAULT.directBuffer(bytes.length);
        decrypt(p, c);
        result = new byte[bytes.length];
        p.getBytes(0, result);
    } finally {
        if (c != null) {
            c.release();
        }
        if (p != null) {
            p.release();
        }
    }
    return result;
}

public byte[] encrypt(byte[] bytes) {
    ByteBuf c = null, p = null;
    byte[] result = null;
    try {
        c = PooledByteBufAllocator.DEFAULT.directBuffer(bytes.length);
        p = Unpooled.wrappedBuffer(bytes);
        encrypt(p, c);
        result = new byte[bytes.length];
        c.getBytes(0, result);
    } finally {
        if (c != null) {
            c.release();
        }
        if (p != null) {
            p.release();
        }
    }
    return result;
}

/**
 * For debugging only.
 * 
 * @param file
 * @return
 * @throws Exception
 */
public static ByteBuf readFile(File file) throws Exception {
    ByteBuf bb = null;
    RandomAccessFile raf = null;
    try {
        raf = new RandomAccessFile(file, "r");
        FileChannel fc = raf.getChannel();
        bb = PooledByteBufAllocator.DEFAULT.directBuffer((int) file.length());
        ByteBuffer buffer = ByteBuffer.allocate(0x1000);
        while (fc.read(buffer) > 0) {
            buffer.flip();
            bb.writeBytes(buffer);
            buffer.clear();
        }
    } catch (Exception e) {
        safeRelease(bb);
        throw e;
    } finally {
        safeClose(raf);
    }
    return bb;
}

@Override
public void initialise(NetworkChannelHandler channelHandler) {
    this.channelHandler = channelHandler;

    final boolean useEpoll = this.configuration.getBoolean("epoll") && Epoll.isAvailable();

    EventLoopGroup acceptGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("acceptGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("acceptGroup"));

    EventLoopGroup ioGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("ioGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("ioGroup"));

    EventLoopGroup channelGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("channelGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("channelGroup"));

    this.serverBootstrap = new ServerBootstrap()
            .group(acceptGroup, ioGroup)
            .channel(useEpoll ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .childHandler(new ChannelInitialiser(channelGroup, this.channelHandler, null))
            .option(ChannelOption.SO_BACKLOG, this.configuration.getInt("backlog"))
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, DefaultMessageSizeEstimator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, this.configuration.getBoolean("tcpNoDelay"))
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
}

private byte[] generateTrailer(ByteBuf buf) {
    VoxelwindHash hash = hashLocal.get();
    ByteBuf counterBuf = PooledByteBufAllocator.DEFAULT.directBuffer(8);
    ByteBuf keyBuf = PooledByteBufAllocator.DEFAULT.directBuffer(serverKey.length);
    try {
        counterBuf.writeLongLE(encryptedSentPacketGenerator.getAndIncrement());
        keyBuf.writeBytes(serverKey);

        hash.update(counterBuf);
        hash.update(buf);
        hash.update(keyBuf);
        byte[] digested = hash.digest();
        return Arrays.copyOf(digested, 8);
    } finally {
        counterBuf.release();
        keyBuf.release();
    }
}

@Override
public void encode(ByteBuf buffer) {
    buffer.writeInt(protocolVersion);
    buffer.writeByte(gameEdition);

    ByteBuf body = PooledByteBufAllocator.DEFAULT.directBuffer();
    try {
        McpeUtil.writeLELengthAsciiString(body, chainData);
        McpeUtil.writeLELengthAsciiString(body, skinData);

        ByteBuf compressed = CompressionUtil.deflate(body);

        Varints.encodeUnsigned(buffer, compressed.readableBytes());
        buffer.writeBytes(compressed);
    } catch (DataFormatException e) {
        throw new RuntimeException("Unable to compress login data body", e);
    } finally {
        body.release();
    }
}

public McpeOverRakNetNetworkListener(VoxelwindServer voxelwindServer, String host, int port, boolean useSoReuseport) {
    this.server = voxelwindServer;
    this.address = new InetSocketAddress(host, port);
    this.useSoReuseport = useSoReuseport;
    if (Epoll.isAvailable()) {
        bootstrap = new Bootstrap()
                .channel(EpollDatagramChannel.class)
                .group(new EpollEventLoopGroup(0, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Voxelwind MCPE Listener - #%d").build()))
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(this);
        if (useSoReuseport) {
            bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
        }
    } else {
        bootstrap = new Bootstrap()
                .channel(NioDatagramChannel.class)
                .group(new NioEventLoopGroup(0, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Voxelwind MCPE Listener - #%d").build()))
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(this);
    }
}

@Override
public boolean bind() {
    ChannelFuture future = new ServerBootstrap()
            .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .group(group)
            .childHandler(this)
            .bind(server.getConfiguration().getRcon().getHost(), server.getConfiguration().getRcon().getPort())
            .awaitUninterruptibly();

    if (future.isSuccess()) {
        this.channel = future.channel();
        return true;
    }

    return false;
}

@Override
public void initChannel(Channel ch) throws Exception
{
    try
    {
        ch.config().setOption( ChannelOption.IP_TOS, 0x18 );
    } catch ( ChannelException ex )
    {
        // IP_TOS is not supported (Windows XP / Windows Server 2003)
    }
    ch.config().setOption( ChannelOption.TCP_NODELAY, true );
    ch.config().setAllocator( PooledByteBufAllocator.DEFAULT );

    ch.pipeline().addLast( TIMEOUT_HANDLER, new ReadTimeoutHandler( BungeeCord.getInstance().config.getTimeout(), TimeUnit.MILLISECONDS ) );
    ch.pipeline().addLast( FRAME_DECODER, new Varint21FrameDecoder() );
    ch.pipeline().addLast( FRAME_PREPENDER, framePrepender );

    ch.pipeline().addLast( BOSS_HANDLER, new HandlerBoss() );
}

/**
 * create a new kcpcb
 *
 * @param output
 * @param user
 */
public Kcp(Output output, Object user)
{
    snd_wnd = IKCP_WND_SND;
    rcv_wnd = IKCP_WND_RCV;
    rmt_wnd = IKCP_WND_RCV;
    mtu = IKCP_MTU_DEF;
    mss = mtu - IKCP_OVERHEAD;
    rx_rto = IKCP_RTO_DEF;
    rx_minrto = IKCP_RTO_MIN;
    interval = IKCP_INTERVAL;
    ts_flush = IKCP_INTERVAL;
    ssthresh = IKCP_THRESH_INIT;
    dead_link = IKCP_DEADLINK;
    buffer = PooledByteBufAllocator.DEFAULT.buffer((mtu + IKCP_OVERHEAD) * 3);
    this.output = output;
    this.user = user;
}

/**
 * change MTU size, default is 1400
 *
 * @param mtu
 * @return
 */
public int setMtu(int mtu)
{
    if (mtu < 50 || mtu < IKCP_OVERHEAD)
    {
        return -1;
    }
    ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer((mtu + IKCP_OVERHEAD) * 3);
    this.mtu = mtu;
    mss = mtu - IKCP_OVERHEAD;
    if (buffer != null)
    {
        buffer.release();
    }
    this.buffer = buf;
    return 0;
}

public static PooledByteBufAllocator createPooledByteBufAllocator(boolean allowDirectBufs,
    boolean allowCache, int numCores) {
  if (numCores == 0) {
    numCores = Runtime.getRuntime().availableProcessors();
  }
  return new PooledByteBufAllocator(allowDirectBufs && PlatformDependent.directBufferPreferred(),
      Math.min(getPrivateStaticField("DEFAULT_NUM_HEAP_ARENA"), numCores), Math.min(
          getPrivateStaticField("DEFAULT_NUM_DIRECT_ARENA"), allowDirectBufs ? numCores : 0),
      getPrivateStaticField("DEFAULT_PAGE_SIZE"), getPrivateStaticField("DEFAULT_MAX_ORDER"),
      allowCache ? getPrivateStaticField("DEFAULT_TINY_CACHE_SIZE") : 0,
      allowCache ? getPrivateStaticField("DEFAULT_SMALL_CACHE_SIZE") : 0,
      allowCache ? getPrivateStaticField("DEFAULT_NORMAL_CACHE_SIZE") : 0);
}