Java 类io.netty.channel.ChannelInboundHandlerAdapter 实例源码

MockClient(EventLoopGroup elg, FrameCodec<ByteBuf> frameCodec) {
  // Set up so written Frames are encoded into bytes, received bytes are encoded into Frames put
  // on queue.
  cb.group(elg)
      .channel(LocalChannel.class)
      .handler(
          new ChannelInitializer<LocalChannel>() {
            @Override
            protected void initChannel(LocalChannel ch) throws Exception {
              ch.pipeline()
                  .addLast(new FrameEncoder(frameCodec))
                  .addLast(new TestFrameDecoder(frameCodec))
                  .addLast(
                      new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg)
                            throws Exception {
                          responses.offer((Frame) msg);
                        }
                      });
            }
          });
}

@Test
public void shouldFireOutboundChannelClosedEvent() throws InterruptedException {
    inboundChannel.pipeline().addLast(handler);

    List<Object> events = new ArrayList<>(1);
    outboundChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            events.add(evt);
        }
    });

    inboundChannel.close().sync();

    assertFalse(events.isEmpty());
    assertEquals(1, events.size());
    assertTrue(events.get(0) instanceof OutboundChannelClosedEvent);
}

@Test
public void serverBootStrapWithOptionsTest() throws InstantiationException, IllegalAccessException, ClassNotFoundException {
    LinkedHashMap<String, Object> channelHandlerOptions = new LinkedHashMap<String, Object>();

    channelHandlerOptions.put("lineFrame", new LineBasedFrameDecoder(2000));
    channelHandlerOptions.put("decoder", new StringDecoder());
    channelHandlerOptions.put("encoder", new StringEncoder());
    channelHandlerOptions.put("handler", new ChannelInboundHandlerAdapter() {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            log.info("Message Received and forward to ConsumerProcessor. Msg -> {}", msg);
        }
    });

    Server server = BootStrap.builder()
            .port(5252)
            .options(channelHandlerOptions)
            .messageConsumer(msg -> log.info(msg))
            .build();

    assertNotNull(server);
}

@Override
protected ChannelHandler newNonSslHandler(ChannelHandlerContext context) {
    return new ChannelInboundHandlerAdapter() {

        private HttpResponseEncoder encoder = new HttpResponseEncoder();

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            LOG.trace("Received non-SSL request, returning redirect");
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.MOVED_PERMANENTLY, Unpooled.EMPTY_BUFFER);
            response.headers().set(Names.LOCATION, redirectAddress);
            LOG.trace(Constants.LOG_RETURNING_RESPONSE, response);
            encoder.write(ctx, response, ctx.voidPromise());
            ctx.flush();
        }
    };
}

public void run() throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();
    try{
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class)
            .remoteAddress(new InetSocketAddress(host, port))
            .handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello DiscardServer.", CharsetUtil.UTF_8));
                        }
                    });
                }
            })
            .option(ChannelOption.SO_KEEPALIVE, true);
        ChannelFuture f = b.connect().sync();
        //
        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

@Override
protected ChannelHandler newNonSslHandler(ChannelHandlerContext context) {
    return new ChannelInboundHandlerAdapter() {

        private HttpResponseEncoder encoder = new HttpResponseEncoder();

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            LOG.trace("Received non-SSL request, returning redirect");
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.MOVED_PERMANENTLY, Unpooled.EMPTY_BUFFER);
            response.headers().set(Names.LOCATION, redirectAddress);
            LOG.trace(Constants.LOG_RETURNING_RESPONSE, response);
            encoder.write(ctx, response, ctx.voidPromise());
            ctx.flush();
        }
    };
}

private void disconnect(ClientSecureChannel secureChannel, CompletableFuture<Unit> disconnected) {
    RequestHeader requestHeader = new RequestHeader(
        NodeId.NULL_VALUE, DateTime.now(), uint(0), uint(0), null, uint(0), null);

    secureChannel.getChannel().pipeline().addLast(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            logger.debug("channelInactive(), disconnect complete");
            disconnected.complete(Unit.VALUE);
            super.channelInactive(ctx);
        }
    });

    logger.debug("Sending CloseSecureChannelRequest...");
    CloseSecureChannelRequest request = new CloseSecureChannelRequest(requestHeader);
    secureChannel.getChannel().pipeline().fireUserEventTriggered(request);

    client.getConfig().getWheelTimer().newTimeout(
        timeout -> disconnected.completeExceptionally(new UaException(StatusCodes.Bad_Timeout)),
        5,
        TimeUnit.SECONDS
    );
}

public static void main(String[] args) throws Exception {
    ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(new PurchaseDataDecoder());
            p.addLast(new PurchaseDataEncoder());
            p.addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx,
                        Object data) throws Exception {
                    System.out.println("processed Purchase " + data);
                    PurchaseData processed = new PurchaseData(data, true);
                    ctx.writeAndFlush(processed);
                }
            });
        }
    };
    BootstrapTemplate.newServerBootstrap(HOST, PORT, initializer);
}

public static void main(String[] args) throws Exception {
    ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(new StringEncoder());
            p.addLast(new StringDecoder());
            p.addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                    System.out.println(msg);
                    ctx.close();
                }
            });
        }
    };
    BootstrapTemplate.newServerBootstrap(HOST, PORT, initializer);
}

public static void main(String[] args) throws Exception {
    ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(new PurchaseDataDecoder());
            p.addLast(new PurchaseDataEncoder());
            p.addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx,
                        Object data) throws Exception {
                    System.out.println("processed Purchase " + data);
                    PurchaseData processed = new PurchaseData(data, true);
                    ctx.writeAndFlush(processed);
                }
            });
        }
    };
    BootstrapTemplate.newServerBootstrap(HOST, PORT, initializer);
}

public static void main(String[] args) throws Exception {
    ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(new StringEncoder());
            p.addLast(new StringDecoder());
            p.addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                    System.out.println(msg);
                    ctx.close();
                }
            });
        }
    };
    BootstrapTemplate.newServerBootstrap(HOST, PORT, initializer);
}

public TestConstantStringServer(int port, final String constant) throws InterruptedException {
    channel = new ServerBootstrap()
            .group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .handler(new LoggingHandler(TestConstantStringServer.class, LogLevel.DEBUG))
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new LoggingHandler(TestConstantStringServer.class, LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            ctx.write(msg);
                            ctx.writeAndFlush(Unpooled.copiedBuffer(constant, CharsetUtil.UTF_8))
                                    .addListener(ChannelFutureListener.CLOSE);
                        }
                    });
                }
            })
            .bind(LOCALHOST, port)
            .sync()
            .channel();
}

@BeforeClass
public static void init() {
    // Configure a test server
    group = new LocalEventLoopGroup();
    ServerBootstrap sb = new ServerBootstrap();
    sb.group(group)
      .channel(LocalServerChannel.class)
      .childHandler(new ChannelInitializer<LocalChannel>() {
          @Override
          public void initChannel(LocalChannel ch) throws Exception {
              ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                  @Override
                  public void channelRead(ChannelHandlerContext ctx, Object msg) {
                      // Discard
                      ReferenceCountUtil.release(msg);
                  }
              });
          }
      });

    localAddr = (LocalAddress) sb.bind(LocalAddress.ANY).syncUninterruptibly().channel().localAddress();
}

@BeforeClass
public static void init() {
    // Configure a test server
    group = new LocalEventLoopGroup();
    ServerBootstrap sb = new ServerBootstrap();
    sb.group(group)
            .channel(LocalServerChannel.class)
            .childHandler(new ChannelInitializer<LocalChannel>() {
                @Override
                public void initChannel(LocalChannel ch) throws Exception {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            // Discard
                            ReferenceCountUtil.release(msg);
                        }
                    });
                }
            });

    localAddr = (LocalAddress) sb.bind(LocalAddress.ANY).syncUninterruptibly().channel().localAddress();
}

@Test
public void testTooManyAcceptedChannels() throws Exception {
    EventLoopGroup g = new OioEventLoopGroup(1);
    ServerBootstrap sb = new ServerBootstrap();
    sb.channel(OioServerSocketChannel.class);
    sb.group(g);
    sb.childHandler(new ChannelInboundHandlerAdapter());
    ChannelFuture f1 = sb.bind(0);
    f1.sync();

    Socket s = new Socket(NetUtil.LOCALHOST, ((InetSocketAddress) f1.channel().localAddress()).getPort());
    assertThat(s.getInputStream().read(), is(-1));
    s.close();

    g.shutdownGracefully();
}

@Test
public void testTcpInfo() throws Exception {
    EventLoopGroup group = new EpollEventLoopGroup(1);

    try {
        Bootstrap bootstrap = new Bootstrap();
        EpollSocketChannel ch = (EpollSocketChannel) bootstrap.group(group)
                .channel(EpollSocketChannel.class)
                .handler(new ChannelInboundHandlerAdapter())
                .bind(new InetSocketAddress(0)).syncUninterruptibly().channel();
        EpollTcpInfo info = ch.tcpInfo();
        assertTcpInfo0(info);
        ch.close().syncUninterruptibly();
    } finally {
        group.shutdownGracefully();
    }
}

@Test
public void testTcpInfoReuse() throws Exception {
    EventLoopGroup group = new EpollEventLoopGroup(1);

    try {
        Bootstrap bootstrap = new Bootstrap();
        EpollSocketChannel ch = (EpollSocketChannel) bootstrap.group(group)
                .channel(EpollSocketChannel.class)
                .handler(new ChannelInboundHandlerAdapter())
                .bind(new InetSocketAddress(0)).syncUninterruptibly().channel();
        EpollTcpInfo info = new EpollTcpInfo();
        ch.tcpInfo(info);
        assertTcpInfo0(info);
        ch.close().syncUninterruptibly();
    } finally {
        group.shutdownGracefully();
    }
}

@Override
public void buildBungeeServer(Channel channel, Connection connection) {
    ChannelPipeline pipeline = channel.pipeline();
    pipeline.addFirst(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(EncapsulatedProtocolUtils.createHandshake(null, false));
            super.channelActive(ctx);
        }
    });
    NetworkDataCache cache = NetworkDataCache.getFrom(connection);
    pipeline.replace(PipelineUtils.PACKET_DECODER, PipelineUtils.PACKET_DECODER, new FromServerPacketDecoder(connection, cache));
    pipeline.replace(PipelineUtils.PACKET_ENCODER, PipelineUtils.PACKET_ENCODER, new ToServerPacketEncoder(connection, cache));
    pipeline.get(CustomHandlerBoss.class).setPacketHandlerChangeListener(listener -> {
        try {
            return (listener instanceof DownstreamBridge) ? new EntityRewriteDownstreamBridge(ProxyServer.getInstance(), ReflectionUtils.getFieldValue(listener, "con")) : listener;
        } catch (IllegalArgumentException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    });
}

@Override
public void buildBungeeServer(Channel channel, Connection connection) {
    ChannelPipeline pipeline = channel.pipeline();
    pipeline.addFirst(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(EncapsulatedProtocolUtils.createHandshake(null, false));
            super.channelActive(ctx);
        }
    });
    NetworkDataCache cache = NetworkDataCache.getFrom(connection);
    pipeline.replace(PipelineUtils.PACKET_DECODER, PipelineUtils.PACKET_DECODER, new FromServerPacketDecoder(connection, cache));
    pipeline.replace(PipelineUtils.PACKET_ENCODER, PipelineUtils.PACKET_ENCODER, new ToServerPacketEncoder(connection, cache));
    pipeline.get(CustomHandlerBoss.class).setPacketHandlerChangeListener(listener -> {
        try {
            return (listener instanceof DownstreamBridge) ? new EntityRewriteDownstreamBridge(ProxyServer.getInstance(), ReflectionUtils.getFieldValue(listener, "con")) : listener;
        } catch (IllegalArgumentException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    });
}

@Override
public void buildBungeeServer(Channel channel, Connection connection) {
    ChannelPipeline pipeline = channel.pipeline();
    pipeline.addFirst(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(EncapsulatedProtocolUtils.createHandshake(null, false));
            super.channelActive(ctx);
        }
    });
    NetworkDataCache cache = NetworkDataCache.getFrom(connection);
    pipeline.replace(PipelineUtils.PACKET_DECODER, PipelineUtils.PACKET_DECODER, new FromServerPacketDecoder(connection, cache));
    pipeline.replace(PipelineUtils.PACKET_ENCODER, PipelineUtils.PACKET_ENCODER, new ToServerPacketEncoder(connection, cache));
    pipeline.get(CustomHandlerBoss.class).setPacketHandlerChangeListener(listener -> {
        try {
            return (listener instanceof DownstreamBridge) ? new EntityRewriteDownstreamBridge(ProxyServer.getInstance(), ReflectionUtils.getFieldValue(listener, "con")) : listener;
        } catch (IllegalArgumentException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    });
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    channel = ch;
    responseEventBus = new ResponseEventBus();
    responseEventBus.getEventBus().register(this);
    ch.pipeline().addLast(new ModbusMasterCodec(transectionId -> {
        Entry entry = requestRecorder.get(transectionId);
        if (null == entry || entry.trieved == true) {
            return null;
        }
        entry.trieved = true;
        return entry.request;
    }), responseEventBus, new ChannelInboundHandlerAdapter() {

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {

            super.channelInactive(ctx);
        }

    });
}

@Test(expected = java.net.BindException.class)
public void portAlreadyInUseTest() {
    final MessageRegistry messageRegistry = new MessageRegistry();
    final NioEventLoopGroup bossGroup = new NioEventLoopGroup();
    final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
    NodeServer serverA = new NodeServer();
    NodeServer serverB = new NodeServer();
    try {
        serverA.initialize("127.0.0.1", 8042, bossGroup, workerGroup, messageRegistry, new ChannelInboundHandlerAdapter() );
        serverB.initialize("127.0.0.1", 8042, bossGroup, workerGroup, messageRegistry, new ChannelInboundHandlerAdapter() );
    } finally {
        serverA.shutdown();
        serverB.shutdown();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

@BeforeClass
public static void init() {
    // Configure a test server
    group = new LocalEventLoopGroup();
    ServerBootstrap sb = new ServerBootstrap();
    sb.group(group)
      .channel(LocalServerChannel.class)
      .childHandler(new ChannelInitializer<LocalChannel>() {
          @Override
          public void initChannel(LocalChannel ch) throws Exception {
              ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                  @Override
                  public void channelRead(ChannelHandlerContext ctx, Object msg) {
                      // Discard
                      ReferenceCountUtil.release(msg);
                  }
              });
          }
      });

    localAddr = (LocalAddress) sb.bind(LocalAddress.ANY).syncUninterruptibly().channel().localAddress();
}

@BeforeClass
public static void init() {
    // Configure a test server
    group = new LocalEventLoopGroup();
    ServerBootstrap sb = new ServerBootstrap();
    sb.group(group)
            .channel(LocalServerChannel.class)
            .childHandler(new ChannelInitializer<LocalChannel>() {
                @Override
                public void initChannel(LocalChannel ch) throws Exception {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            // Discard
                            ReferenceCountUtil.release(msg);
                        }
                    });
                }
            });

    localAddr = (LocalAddress) sb.bind(LocalAddress.ANY).syncUninterruptibly().channel().localAddress();
}

@Test
public void testTooManyAcceptedChannels() throws Exception {
    EventLoopGroup g = new OioEventLoopGroup(1);
    ServerBootstrap sb = new ServerBootstrap();
    sb.channel(OioServerSocketChannel.class);
    sb.group(g);
    sb.childHandler(new ChannelInboundHandlerAdapter());
    ChannelFuture f1 = sb.bind(0);
    f1.sync();

    Socket s = new Socket(NetUtil.LOCALHOST, ((InetSocketAddress) f1.channel().localAddress()).getPort());
    assertThat(s.getInputStream().read(), is(-1));
    s.close();

    g.shutdownGracefully();
}

@Test(expected = java.net.BindException.class)
public void portAlreadyInUseTest() {
    final MessageRegistry messageRegistry = new MessageRegistry();
    final NioEventLoopGroup bossGroup = new NioEventLoopGroup();
    final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
    NodeServer serverA = new NodeServer();
    NodeServer serverB = new NodeServer();
    try {
        serverA.initialize("127.0.0.1", 8042, bossGroup, workerGroup, messageRegistry, new ChannelInboundHandlerAdapter() );
        serverB.initialize("127.0.0.1", 8042, bossGroup, workerGroup, messageRegistry, new ChannelInboundHandlerAdapter() );
    } finally {
        serverA.shutdown();
        serverB.shutdown();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

private void disconnect(ClientSecureChannel secureChannel, CompletableFuture<Void> disconnected) {
    RequestHeader requestHeader = new RequestHeader(
        NodeId.NULL_VALUE, DateTime.now(), uint(0), uint(0), null, uint(0), null);

    secureChannel.getChannel().pipeline().addFirst(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            logger.debug("channelInactive(), disconnect complete");
            disconnected.complete(null);
        }
    });

    logger.debug("Sending CloseSecureChannelRequest...");
    CloseSecureChannelRequest request = new CloseSecureChannelRequest(requestHeader);
    secureChannel.getChannel().pipeline().fireUserEventTriggered(request);
}

/**
 * First let's just exercise the PipelineTester a bit.
 */
@Test
public void testPipelineTester() {
  final ByteBuf buf = Unpooled.copiedBuffer("Hello, world", UTF_8);

  final PipelineTester pipelineTester = new PipelineTester(new ChannelInboundHandlerAdapter() {

    @Override
    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
      super.channelActive(ctx);
      ctx.channel().writeAndFlush(buf);
    }
  });
  assertThat(pipelineTester.readClient(), is(buf));

  final ByteBuf foo = Unpooled.copiedBuffer("foo", UTF_8);
  pipelineTester.writeClient(foo.retain());

  assertThat(foo, is(pipelineTester.readServer()));

  final ByteBuf bar = Unpooled.copiedBuffer("bar", UTF_8);
  pipelineTester.writeServer(bar.retain());
  assertThat(bar, is(pipelineTester.readClient()));
}

@Scope("prototype")
    @Bean(name = "channelInitializer")
    public ChannelInboundHandlerAdapter channelInitializer() throws Exception {

//        return new ChannelInitializer<SocketChannel>() {
//            @Override
//            public void initChannel(SocketChannel ch) throws Exception {
//                ChannelPipeline pipeline = ch.pipeline();
//
//                boolean loggingHadler = java.lang.management.ManagementFactory.getRuntimeMXBean().getInputArguments().toString().indexOf("-agentlib:jdwp") > 0 || isDebug;
//
//                if (loggingHadler) {
//                    pipeline.addLast("logger", new LoggingHandler());
//                }
//
//                ProtocolUnificationHandler protocolUnificationHandler = new ProtocolUnificationHandler(isUnificationMode, opcodeMap(), sslContext(), maxFrameLength, charset);
//                protocolUnificationHandler.setRedisService(redisService);
//                pipeline.addLast(protocolUnificationHandler);
//            }
//        };

        ChannelServerInitializer channelServerInitializer = new ChannelServerInitializer();
        channelServerInitializer.setOpcodeMap(opcodeMap());
        channelServerInitializer.setRedisService(redisService);
        channelServerInitializer.setCharset(charset);
        channelServerInitializer.setSslCtx(sslContext());
        channelServerInitializer.setMaxFrameLength(8192);

        return channelServerInitializer;
    }

public NettyWebSocketServer(String host,int port,String uri,ChannelInboundHandlerAdapter handler) {
    super(new InetSocketAddress(host, port), handler);
    if(uri==null || uri.isEmpty())
    {
        this.uri="/";
    }else
    {
        this.uri=uri;
    }
}

public NettyWebSocketServer(NettyServerConfig config,String host,int port,String uri,ChannelInboundHandlerAdapter handler) {
    super(config,new InetSocketAddress(host, port), handler);
    if(uri==null || uri.isEmpty())
    {
        this.uri="/";
    }else
    {
        this.uri=uri;
    }
}

public void server(int port) throws Exception {
    final ByteBuf buf = Unpooled.unreleasableBuffer(
            Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();

        b.group(group)
         .channel(OioServerSocketChannel.class)
         .localAddress(new InetSocketAddress(port))
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) 
                 throws Exception {
                 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                     @Override
                     public void channelActive(ChannelHandlerContext ctx) throws Exception {
                         ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);
                     }
                 });
             }
         });
        ChannelFuture f = b.bind().sync();
        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully().sync();
    }
}

/**
 * Initializes this socket and binds its internal udp socket to a free port.
 * If the socket is already initialized any invocation of this method will
 * result in an IllegalStateException.
 *
 * @throws SocketException Thrown in case the socket could not be initialized
 */
public void initialize() throws SocketException {
    if ( this.isInitialized() ) {
        throw new IllegalStateException( "Cannot re-initialized ClientSocket" );
    }

    this.udpSocket = new Bootstrap();
    this.udpSocket.group( Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup() );
    this.udpSocket.channel( Epoll.isAvailable() ? EpollDatagramChannel.class : NioDatagramChannel.class );
    this.udpSocket.handler( new ChannelInboundHandlerAdapter() {
        @Override
        public void channelRead( ChannelHandlerContext ctx, Object msg ) throws Exception {
            io.netty.channel.socket.DatagramPacket packet = (io.netty.channel.socket.DatagramPacket) msg;
            PacketBuffer content = new PacketBuffer( packet.content() );
            InetSocketAddress sender = packet.sender();

            if ( !receiveDatagram( sender, content ) ) {
                // Push datagram to update queue:
                handleDatagram( sender, content, System.currentTimeMillis() );
            }
        }
    } );

    try {
        this.channel = this.udpSocket.bind( ThreadLocalRandom.current().nextInt( 45000, 65000 ) ).sync().channel();
    } catch ( InterruptedException e ) {
        SocketException exception = new SocketException( "Could not bind to socket" );
        exception.initCause( e );
        throw exception;
    }

    this.afterInitialize();
}

@Test
public void afterNettyContextInit() {
    AtomicInteger readCount = new AtomicInteger();
    ChannelInboundHandlerAdapter handler = new ChannelInboundHandlerAdapter() {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            readCount.incrementAndGet();
            super.channelRead(ctx, msg);
        }
    };
    String handlerName = "test";

    NettyContext nettyContext =
            HttpServer.create(opt -> opt.afterNettyContextInit(c -> c.addHandlerFirst(handlerName, handler)))
                      .start((req, resp) -> resp.sendNotFound())
                      .getContext();

    HttpClientResponse response1 = HttpClient.create(opt -> opt.connectAddress(() -> nettyContext.address()))
                                             .get("/", req -> req.failOnClientError(false).send())
                                             .block();

    assertThat(response1.status().code()).isEqualTo(404);
    response1.dispose();

    //the "main" context doesn't get enriched with handlers from options...
    assertThat(nettyContext.channel().pipeline().names()).doesNotContain(handlerName);
    //...but the child channels that are created for requests are
    assertThat(readCount.get()).isEqualTo(1);

    HttpClientResponse response2 = HttpClient.create(opt -> opt.connectAddress(() -> nettyContext.address()))
                                             .get("/", req -> req.failOnClientError(false).send())
                                             .block();

    assertThat(response2.status().code()).isEqualTo(404); //reactor handler was applied and produced a response
    response2.dispose();
    assertThat(readCount.get()).isEqualTo(1); //BUT channelHandler wasn't applied a second time since not Shareable

    nettyContext.dispose();
}

private void createServerChannelHandler() {
    endInitProtocol = new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel channel) throws Exception {
            try {
                synchronized (networkManagers) {
                    if (!closed) 
                        injectChannelInternal(channel);
                }
            } 
            catch (Exception e) {
                plugin.getLogger().log(Level.SEVERE, "Cannot inject incomming channel " + channel, e);
            }
        }

    };

    beginInitProtocol = new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel channel) throws Exception {
            channel.pipeline().addLast(endInitProtocol);
        }

    };

    serverChannelHandler = new ChannelInboundHandlerAdapter() {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            Channel channel = (Channel) msg;

            channel.pipeline().addFirst(beginInitProtocol);
            ctx.fireChannelRead(msg);
        }

    };
}

@Override
public void prepare() throws Exception {

    eventHandleAdaptor = new ChannelInboundHandlerAdapter() {

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

            //              System.out.println("_________________"+msg);

            //              ctx.write(msg);

            addCount(1);
        }
    };

    Bootstrap b = new Bootstrap();
    b.group(group);
    b.channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, false);
    b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("frameDecoder",
                    new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
            pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));

            pipeline.addLast("handler", eventHandleAdaptor);
        }
    });

    f = b.connect("localhost", 5656).sync();
}

public void start(TcpChannelHandler handler) {  
    channelHandler = new ChannelInboundHandlerAdapter() {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)  throws Exception {
            handler.process(ctx, msg);
        }
    };      
    start();
}

public static void main(String[] args) {
//      new SimpleTcpServer(8007).start( (ChannelHandlerContext ctx, Object msg) -> {
//          System.out.println(msg);
//          ctx.writeAndFlush("ok");
//      });
        new SimpleTcpServer(8007).start(new ChannelInboundHandlerAdapter(){
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg)
                    throws Exception {
                ctx.writeAndFlush("ok");
            }

        });
    }

@Override
public final ChannelInitializer<Channel> initializer() {
    return new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            final ChannelInboundHandlerAdapter exceptionHandler =
                new ChannelInboundHandlerAdapter() {
                    @Override
                    public void exceptionCaught(
                        final ChannelHandlerContext ctx, final Throwable cause
                    ) throws Exception {
                        if (cause instanceof HttpException) {
                            final HttpException e = (HttpException) cause;
                            sendResponse(ctx, e.getStatus());
                            return;
                        }

                        if (cause instanceof DecoderException) {
                            exceptionCaught(ctx, cause.getCause());
                            return;
                        }

                        log.error("error in pipeline: ", cause);
                        sendResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR);
                    }
                };
            ch
                .pipeline()
                .addLast(new HttpRequestDecoder(), new HttpContentDecompressor(),
                    new HttpObjectAggregator(Integer.MAX_VALUE), decoder, exceptionHandler,
                    handler);
            ch.pipeline().addLast(new HttpResponseEncoder());
        }
    };
}

static ChannelHandler forbiddenHttpRequestResponder() {
    return new ChannelInboundHandlerAdapter() {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof FullHttpRequest) {
                ((FullHttpRequest) msg).release();
                FullHttpResponse response =
                        new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.FORBIDDEN);
                ctx.channel().writeAndFlush(response);
            } else {
                ctx.fireChannelRead(msg);
            }
        }
    };
}