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

@Test
public void addByteDecoderWhenNoLeft() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerLast("decoder", decoder)
               .addHandlerFirst("decoder$extract",
                       NettyPipeline.inboundHandler(ADD_EXTRACTOR));

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList("decoder$extract",
                    "decoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteDecoderWhenNoRight() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerLast("decoder", decoder)
               .addHandlerFirst("decoder$extract",
                       NettyPipeline.inboundHandler(ADD_EXTRACTOR));

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    "decoder$extract",
                    "decoder",
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteDecoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerLast("decoder", decoder)
               .addHandlerFirst("decoder$extract",
                       NettyPipeline.inboundHandler(ADD_EXTRACTOR));

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "decoder$extract",
                    "decoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteDecoderWhenNoLeft() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerLast("decoder", decoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList("decoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteDecoderWhenNoRight() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerLast("decoder", decoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    "decoder",
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteDecoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerLast("decoder", decoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "decoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhenNoLeft() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList("encoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhenNoRight() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    "encoder",
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "encoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteEncoderWhenNoLeft() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList("encoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteEncoderWhenNoRight() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    "encoder",
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteEncoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "encoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addSeveralByteEncodersWhenCodec() throws Exception {
    ChannelHandler encoder1 = new LineBasedFrameDecoder(12);
    ChannelHandler encoder2 = new LineBasedFrameDecoder(13);

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });

    testContext.addHandlerFirst("encoder1", encoder1)
               .addHandlerFirst("encoder2", encoder2);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "encoder2",
                    "encoder1",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void testConstructorWithProvidedReplacement() {
    EmbeddedChannel channel = new EmbeddedChannel();
    channel.pipeline().addFirst(NettyPipeline.SslHandler, new ChannelHandlerAdapter() {
    });

    HttpClientOperations ops1 = new HttpClientOperations(channel,
            (response, request) -> null, handler);
    ops1.followRedirect();
    ops1.failOnClientError(false);
    ops1.failOnServerError(false);

    HttpClientOperations ops2 = new HttpClientOperations(channel, ops1);

    assertSame(ops1.channel(), ops2.channel());
    assertSame(ops1.started, ops2.started);
    assertSame(ops1.redirectedFrom, ops2.redirectedFrom);
    assertSame(ops1.isSecure, ops2.isSecure);
    assertSame(ops1.nettyRequest, ops2.nettyRequest);
    assertSame(ops1.responseState, ops2.responseState);
    assertSame(ops1.redirectable, ops2.redirectable);
    assertSame(ops1.inboundPrefetch, ops2.inboundPrefetch);
    assertSame(ops1.requestHeaders, ops2.requestHeaders);
    assertSame(ops1.clientError, ops2.clientError);
    assertSame(ops1.serverError, ops2.serverError);
}

public void run(String host, int port) throws InterruptedException {
    NioEventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelHandlerAdapter() {
                            @Override
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                System.out.println("connection active");
                            }
                        });
                    }
                });
        ChannelFuture future = bootstrap.connect(host, port).sync();
        future.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public void testConnectCancellation(Bootstrap cb) throws Throwable {
    cb.handler(new ChannelHandlerAdapter()).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000);
    ChannelFuture future = cb.connect(BAD_HOST, 8080);
    try {
        if (future.await(1000)) {
            if (future.isSuccess()) {
                fail("A connection attempt to " + BAD_HOST + " must not succeed.");
            } else {
                throw future.cause();
            }
        }

        if (future.cancel(true)) {
            assertThat(future.channel().closeFuture().await(500), is(true));
            assertThat(future.isCancelled(), is(true));
        } else {
            // Cancellation not supported by the transport.
        }
    } finally {
        future.channel().close();
    }
}

@BeforeClass
public static void init() {
    // Configure a test server
    group = new DefaultEventLoopGroup();
    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 ChannelHandlerAdapter() {
                  @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 DefaultEventLoopGroup();
    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 ChannelHandlerAdapter() {
                        @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 ChannelHandlerAdapter());
    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();
}

/**
 * Create a server plaintext handler for gRPC.
 */
public static ProtocolNegotiator serverPlaintext() {
  return new ProtocolNegotiator() {
    @Override
    public Handler newHandler(final GrpcHttp2ConnectionHandler handler) {
      class PlaintextHandler extends ChannelHandlerAdapter implements Handler {
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
          // Set sttributes before replace to be sure we pass it before accepting any requests.
          handler.handleProtocolNegotiationCompleted(Attributes.newBuilder()
              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
              .build());
          // Just replace this handler with the gRPC handler.
          ctx.pipeline().replace(this, null, handler);
        }

        @Override
        public AsciiString scheme() {
          return Utils.HTTP;
        }
      }

      return new PlaintextHandler();
    }
  };
}

@Before
public void setUp() throws Exception {
    master = mock(NitmProxyMaster.class);
    when(master.config()).thenReturn(new NitmProxyConfig());
    when(master.handler(any(), any(), any())).thenAnswer(m -> new ChannelHandlerAdapter() {
    });

    inboundChannel = new EmbeddedChannel();
}

@Test
public void addNonByteDecoderWhenEmptyPipeline() throws Exception {

    ChannelHandler decoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerLast("decoder", decoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList("decoder", "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addSeveralByteDecodersWhenCodec() throws Exception {
    ChannelHandler decoder1 = new LineBasedFrameDecoder(12);
    ChannelHandler decoder2 = new LineBasedFrameDecoder(13);

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });

    testContext.addHandlerLast("decoder1$extract",
            NettyPipeline.inboundHandler(ADD_EXTRACTOR))
               .addHandlerLast("decoder1", decoder1)

               .addHandlerLast("decoder2$extract",
                       NettyPipeline.inboundHandler(ADD_EXTRACTOR))
               .addHandlerLast("decoder2", decoder2);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "decoder1$extract",
                    "decoder1",
                    "decoder2$extract",
                    "decoder2",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteEncoderWhenEmptyPipeline() throws Exception {

    ChannelHandler encoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList("encoder", "DefaultChannelPipeline$TailContext#0"));
}

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

public void testConnectTimeout(Bootstrap cb) throws Throwable {
    cb.handler(new ChannelHandlerAdapter()).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000);
    ChannelFuture future = cb.connect(BAD_HOST, 8080);
    try {
        assertThat(future.await(3000), is(true));
    } finally {
        future.channel().close();
    }
}

@Test
public void testNotThrowBlockingOperationException() throws Exception {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    final ChannelGroup allChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup);
    b.childHandler(new ChannelHandlerAdapter() {
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            allChannels.add(ctx.channel());
        }
    });
    b.channel(NioServerSocketChannel.class);

    ChannelFuture f = b.bind(0).syncUninterruptibly();

    if (f.isSuccess()) {
        allChannels.add(f.channel());
        allChannels.close().awaitUninterruptibly();
    }

    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
    bossGroup.terminationFuture().sync();
    workerGroup.terminationFuture().sync();
}

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

    ChannelFuture f2 = b.bind(0);
    f2.await();

    assertThat(f2.cause(), is(instanceOf(ChannelException.class)));
    assertThat(f2.cause().getMessage().toLowerCase(), containsString("too many channels"));

    final CountDownLatch notified = new CountDownLatch(1);
    f2.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            notified.countDown();
        }
    });

    notified.await();
    g.shutdownGracefully();
}

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

    Bootstrap cb = new Bootstrap();
    cb.channel(OioSocketChannel.class);
    cb.group(g);
    cb.handler(new ChannelHandlerAdapter());
    ChannelFuture f2 = cb.connect(NetUtil.LOCALHOST, ((InetSocketAddress) f1.channel().localAddress()).getPort());
    f2.await();

    assertThat(f2.cause(), is(instanceOf(ChannelException.class)));
    assertThat(f2.cause().getMessage().toLowerCase(), containsString("too many channels"));

    final CountDownLatch notified = new CountDownLatch(1);
    f2.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            notified.countDown();
        }
    });

    notified.await();
    g.shutdownGracefully();
}

/**
 * Reproduces the issue #1679
 */
@Test
public void testFlushAfterGatheredFlush() throws Exception {
    NioEventLoopGroup group = new NioEventLoopGroup(1);
    try {
        ServerBootstrap sb = new ServerBootstrap();
        sb.group(group).channel(NioServerSocketChannel.class);
        sb.childHandler(new ChannelHandlerAdapter() {
            @Override
            public void channelActive(final ChannelHandlerContext ctx) throws Exception {
                // Trigger a gathering write by writing two buffers.
                ctx.write(Unpooled.wrappedBuffer(new byte[] { 'a' }));
                ChannelFuture f = ctx.write(Unpooled.wrappedBuffer(new byte[] { 'b' }));
                f.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        // This message must be flushed
                        ctx.writeAndFlush(Unpooled.wrappedBuffer(new byte[]{'c'}));
                    }
                });
                ctx.flush();
            }
        });

        SocketAddress address = sb.bind(0).sync().channel().localAddress();

        Socket s = new Socket();
        s.connect(address);

        DataInput in = new DataInputStream(s.getInputStream());
        byte[] buf = new byte[3];
        in.readFully(buf);

        assertThat(new String(buf, CharsetUtil.US_ASCII), is("abc"));

        s.close();
    } finally {
        group.shutdownGracefully().sync();
    }
}

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

public static void main(String[] args) throws IOException, InterruptedException {
    Bootstrap b = new Bootstrap();
    b.group(new NioEventLoopGroup()).channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<NioSocketChannel>() {

            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast("dd", new ChannelHandlerAdapter() {

                    /**
                     * {@inheritDoc}
                     * 
                     * @see io.netty.channel.ChannelHandlerAdapter#exceptionCaught(io.netty.channel.ChannelHandlerContext,
                     *      java.lang.Throwable)
                     */
                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                        System.out.println("error: ");
                        cause.printStackTrace();
                    }
                });
            }
        });
    b.connect("localhost", 8090).addListener(new ChannelFutureListener() {

        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                future.channel().write(Unpooled.buffer().writeBytes("123".getBytes()));
                future.channel().flush();
            }
        }
    });
}

@Override
public Handler newHandler(GrpcHttp2ConnectionHandler handler) {
  ChannelHandler sslBootstrap = new ChannelHandlerAdapter() {
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
      SSLEngine sslEngine = sslContext.newEngine(ctx.alloc(), host, port);
      SSLParameters sslParams = sslEngine.getSSLParameters();
      sslParams.setEndpointIdentificationAlgorithm("HTTPS");
      sslEngine.setSSLParameters(sslParams);
      ctx.pipeline().replace(this, null, new SslHandler(sslEngine, false));
    }
  };
  return new BufferUntilTlsNegotiatedHandler(sslBootstrap, handler);
}

/**
 * Throws {@link IllegalStateException} if {@link ChannelHandlerAdapter#isSharable()} returns {@code true}
 */
static void ensureNotSharable(ChannelHandlerAdapter handler) {
    if (handler.isSharable()) {
        throw new IllegalStateException("@Sharable annotation is not allowed");
    }
}

/**
 * @param sslContext SslContext context used to build an SslHandler
 * @param cleartextHandler ChannelHandlerAdapter handler attached to the pipeline when a cleartext
 *     connection is received; must be Sharable
 */
public GentleSslHandler(SslContext sslContext, ChannelHandlerAdapter cleartextHandler) {
  super(sslContext);
  Preconditions.checkArgument(cleartextHandler.isSharable(), "cleartextHandler must be Sharable");
  this.cleartextHandler = cleartextHandler;
}

/**
 * Reproduces the issue #1600
 */
@Test
public void testFlushCloseReentrance() throws Exception {
    NioEventLoopGroup group = new NioEventLoopGroup(1);
    try {
        final Queue<ChannelFuture> futures = new LinkedBlockingQueue<ChannelFuture>();

        ServerBootstrap sb = new ServerBootstrap();
        sb.group(group).channel(NioServerSocketChannel.class);
        sb.childHandler(new ChannelHandlerAdapter() {
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                // Write a large enough data so that it is split into two loops.
                futures.add(ctx.write(
                        ctx.alloc().buffer().writeZero(1048576)).addListener(ChannelFutureListener.CLOSE));
                futures.add(ctx.write(ctx.alloc().buffer().writeZero(1048576)));
                ctx.flush();
                futures.add(ctx.write(ctx.alloc().buffer().writeZero(1048576)));
                ctx.flush();
            }
        });

        SocketAddress address = sb.bind(0).sync().channel().localAddress();

        Socket s = new Socket();
        s.connect(address);

        InputStream in = s.getInputStream();
        byte[] buf = new byte[8192];
        for (;;) {
            if (in.read(buf) == -1) {
                break;
            }

            // Wait a little bit so that the write attempts are split into multiple flush attempts.
            Thread.sleep(10);
        }
        s.close();

        assertThat(futures.size(), is(3));
        ChannelFuture f1 = futures.poll();
        ChannelFuture f2 = futures.poll();
        ChannelFuture f3 = futures.poll();
        assertThat(f1.isSuccess(), is(true));
        assertThat(f2.isDone(), is(true));
        assertThat(f2.isSuccess(), is(false));
        assertThat(f2.cause(), is(instanceOf(ClosedChannelException.class)));
        assertThat(f3.isDone(), is(true));
        assertThat(f3.isSuccess(), is(false));
        assertThat(f3.cause(), is(instanceOf(ClosedChannelException.class)));
    } finally {
        group.shutdownGracefully().sync();
    }
}

/**
     * Launches the server.
     * @param port the port.
     * @param handler the {@link ChannelHandlerAdapter handler} to handle incoming connections.
     * @return the {@link ChannelFuture}.
     * @throws InterruptedException if the server gets interrupted.
     * @throws IOException if the socket cannot be bound.
     */
    @NotNull
    protected ChannelFuture launchServer(final int port, @NotNull final ChannelHandlerAdapter handler)
        throws InterruptedException,
               IOException
    {
        @NotNull final ChannelFuture result;

        @Nullable ChannelFuture aux = null;

        @NotNull final EventLoopGroup bossGroup = new NioEventLoopGroup();
        setEventLoopGroup(bossGroup);
        @NotNull final EventLoopGroup workerGroup = new NioEventLoopGroup();
        try
        {
            @NotNull final ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>()
                { // (4)
                    /**
                     * {@inheritDoc}
                     */
                    @Override
                    public void initChannel(@NotNull final SocketChannel ch)
                        throws Exception
                    {
                        ch.pipeline().addLast(handler);
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)          // (5)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
                .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)

            // Bind and start to accept incoming connections.
            aux = b.bind(port).sync(); // (7)

            // Wait until the server socket is closed.
            // In this example, this does not happen, but you can do that to gracefully
            // shut down your server.
//            result.channel().closeFuture().sync();
        }
        catch (@NotNull final Throwable throwable)
        {
            LogFactory.getLog(NettyServerDebuggingService.class).fatal(
                "Cannot run the template debugging server", throwable);
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }

        if (aux == null)
        {
            throw new RuntimeException("Cannot run server");
        }
        else
        {
            result = aux;
        }

        return result;
    }