java类io.netty.channel.local.LocalAddress的实例源码

@Test
public void shouldInheritClusterOverride() {
  BoundCluster cluster =
      new BoundCluster(
          ClusterSpec.builder().withPeerInfo("protocol_versions", Lists.newArrayList(5)).build(),
          0L,
          null);
  BoundDataCenter dc = new BoundDataCenter(cluster);
  BoundNode node =
      new BoundNode(
          new LocalAddress(UUID.randomUUID().toString()),
          NodeSpec.builder().withName("node0").withId(0L).build(),
          Collections.emptyMap(),
          cluster,
          dc,
          null,
          timer,
          null, // channel reference only needed for closing, not useful in context of this test.
          false);

  assertThat(node.getFrameCodec().getSupportedProtocolVersions()).containsOnly(5);
  assertThat(dc.getFrameCodec().getSupportedProtocolVersions()).containsOnly(5);
  assertThat(cluster.getFrameCodec().getSupportedProtocolVersions()).containsOnly(5);
}

@Test
public void shouldInheritClusterOverrideFromCassandraVersion() {
  BoundCluster cluster =
      new BoundCluster(ClusterSpec.builder().withCassandraVersion("2.1.17").build(), 0L, null);
  BoundDataCenter dc = new BoundDataCenter(cluster);
  BoundNode node =
      new BoundNode(
          new LocalAddress(UUID.randomUUID().toString()),
          NodeSpec.builder().withName("node0").withId(0L).build(),
          Collections.emptyMap(),
          cluster,
          dc,
          null,
          timer,
          null, // channel reference only needed for closing, not useful in context of this test.
          false);

  assertThat(node.getFrameCodec().getSupportedProtocolVersions()).containsOnly(3);
  assertThat(dc.getFrameCodec().getSupportedProtocolVersions()).containsOnly(3);
  assertThat(cluster.getFrameCodec().getSupportedProtocolVersions()).containsOnly(3);
}

@Test
public void testShouldUseProtocolVersionOverride() {
  BoundCluster cluster = new BoundCluster(ClusterSpec.builder().build(), 0L, null);
  BoundDataCenter dc = new BoundDataCenter(cluster);
  BoundNode node =
      new BoundNode(
          new LocalAddress(UUID.randomUUID().toString()),
          NodeSpec.builder()
              .withName("node0")
              .withId(0L)
              .withCassandraVersion("2.1.17")
              .withPeerInfo("protocol_versions", Lists.newArrayList(4))
              .build(),
          Collections.emptyMap(),
          cluster,
          dc,
          null,
          timer,
          null, // channel reference only needed for closing, not useful in context of this test.
          false);

  assertThat(node.getFrameCodec().getSupportedProtocolVersions()).containsOnly(4);
}

public void testProtocolVersionForCassandraVersion(
    String cassandraVersion, Integer... expectedProtocolVersions) {
  BoundCluster cluster = new BoundCluster(ClusterSpec.builder().build(), 0L, null);
  BoundDataCenter dc = new BoundDataCenter(cluster);
  BoundNode node =
      new BoundNode(
          new LocalAddress(UUID.randomUUID().toString()),
          NodeSpec.builder()
              .withName("node0")
              .withId(0L)
              .withCassandraVersion(cassandraVersion)
              .build(),
          Collections.emptyMap(),
          cluster,
          dc,
          null,
          timer,
          null, // channel reference only needed for closing, not useful in context of this test.
          false);

  assertThat(node.getFrameCodec().getSupportedProtocolVersions())
      .containsOnly(expectedProtocolVersions);
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

public static void main(String[] args) throws Exception {
    NettyServerBuilder.forAddress(LocalAddress.ANY).forPort(19876)
        .maxConcurrentCallsPerConnection(12).maxMessageSize(16777216)
        .addService(new MockApplicationRegisterService())
        .addService(new MockInstanceDiscoveryService())
        .addService(new MockJVMMetricsService())
        .addService(new MockServiceNameDiscoveryService())
        .addService(new MockTraceSegmentService()).build().start();

    Server jettyServer = new Server(new InetSocketAddress("0.0.0.0",
        Integer.valueOf(12800)));
    String contextPath = "/";
    ServletContextHandler servletContextHandler = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
    servletContextHandler.setContextPath(contextPath);
    servletContextHandler.addServlet(GrpcAddressHttpService.class, GrpcAddressHttpService.SERVLET_PATH);
    servletContextHandler.addServlet(ReceiveDataService.class, ReceiveDataService.SERVLET_PATH);
    servletContextHandler.addServlet(ClearReceiveDataService.class, ClearReceiveDataService.SERVLET_PATH);
    jettyServer.setHandler(servletContextHandler);
    jettyServer.start();
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)SERVER_NIO_EVENTLOOP.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * Adds a channel that listens locally
 */
@SideOnly(Side.CLIENT)
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)SERVER_NIO_EVENTLOOP.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

public SimpleConnectionPool(Bootstrap bootstrap, HandlerConfig handlerConfig, SocketAddress remoteAddress, int connectTimeout) {
    super(bootstrap, new RpcClientChannelPoolHandler(handlerConfig, remoteAddress));
    this.connectTimeout = connectTimeout;
    this.socketAddress = remoteAddress;
    if (remoteAddress instanceof InetSocketAddress) {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) remoteAddress;
        host = inetSocketAddress.getAddress().getHostAddress();
        port = inetSocketAddress.getPort();
    } else if (remoteAddress instanceof LocalAddress) {
        LocalAddress localAddress = (LocalAddress) remoteAddress;
        int myPort = -1;
        try {
            myPort = Integer.parseInt(localAddress.id());
        } catch (NumberFormatException e) {
               throw new RpcFrameworkException(localAddress.id() + " port parse error", e);
        }

        host = "local";
        port = myPort;
    } else {
           throw new RpcFrameworkException(
                "SocketAddress must be '" + InetSocketAddress.class.getName() + "' or '" + LocalAddress.class.getName() + "' (sub) class");
    }

    poolContext = new ConnectionPoolContext(handlerConfig.getResponsePromiseContainer());
}

/**
 * Sets up a server channel bound to the given local address.
 *
 * @return the event loop group used to process incoming connections.
 */
static EventLoopGroup setUpServer(
    ChannelInitializer<LocalChannel> serverInitializer, LocalAddress localAddress)
    throws Exception {
  // Only use one thread in the event loop group. The same event loop group will be used to
  // register client channels during setUpClient as well. This ensures that all I/O activities
  // in both channels happen in the same thread, making debugging easier (i. e. no need to jump
  // between threads when debugging, everything happens synchronously within the only I/O
  // effectively). Note that the main thread is still separate from the I/O thread and
  // synchronization (using the lock field) is still needed when the main thread needs to verify
  // properties calculated by the I/O thread.
  EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
  ServerBootstrap sb =
      new ServerBootstrap()
          .group(eventLoopGroup)
          .channel(LocalServerChannel.class)
          .childHandler(serverInitializer);
  ChannelFuture unusedFuture = sb.bind(localAddress).syncUninterruptibly();
  return eventLoopGroup;
}

/**
 * Adds a channel that listens locally
 */
@SideOnly(Side.CLIENT)
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)SERVER_NIO_EVENTLOOP.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

@SideOnly(Side.CLIENT)
public SocketAddress addLocalEndpoint()
{
    List list = this.endpoints;
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer()
        {
            private static final String __OBFID = "CL_00001449";
            protected void initChannel(Channel p_initChannel_1_)
            {
                NetworkManager networkmanager = new NetworkManager(false);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast("packet_handler", networkmanager);
            }
        }).group(eventLoops).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

@SideOnly(Side.CLIENT)
public SocketAddress addLocalEndpoint()
{
    List list = this.endpoints;
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer()
        {
            private static final String __OBFID = "CL_00001449";
            protected void initChannel(Channel p_initChannel_1_)
            {
                NetworkManager networkmanager = new NetworkManager(false);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast("packet_handler", networkmanager);
            }
        }).group(eventLoops).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

@Test
public void shouldInheritDCOverride() {
  ClusterSpec clusterSpec =
      ClusterSpec.builder().withPeerInfo("protocol_versions", Lists.newArrayList(5)).build();
  BoundCluster cluster = new BoundCluster(clusterSpec, 0L, null);
  DataCenterSpec dcSpec =
      clusterSpec
          .addDataCenter()
          .withPeerInfo("protocol_versions", Lists.newArrayList(4))
          .build();
  BoundDataCenter dc = new BoundDataCenter(dcSpec, cluster);
  BoundNode node =
      new BoundNode(
          new LocalAddress(UUID.randomUUID().toString()),
          NodeSpec.builder().withName("node0").withId(0L).build(),
          Collections.emptyMap(),
          cluster,
          dc,
          null,
          timer,
          null, // channel reference only needed for closing, not useful in context of this test.
          false);

  assertThat(node.getFrameCodec().getSupportedProtocolVersions()).containsOnly(4);
  assertThat(dc.getFrameCodec().getSupportedProtocolVersions()).containsOnly(4);
  assertThat(cluster.getFrameCodec().getSupportedProtocolVersions()).containsOnly(5);
}

public static String getHostString(SocketAddress address) {
    if (address instanceof InetSocketAddress) {
        return ((InetSocketAddress) address).getHostString();
    } else if (address instanceof LocalAddress) {
        return LOCAL_ADDRESS;
    }

    return address.toString();
}

@Test
public void testFailure_defaultTrustManager_rejectSelfSignedCert() throws Exception {
  SelfSignedCertificate ssc = new SelfSignedCertificate(SSL_HOST);
  LocalAddress localAddress = new LocalAddress("DEFAULT_TRUST_MANAGER_REJECT_SELF_SIGNED_CERT");
  Lock clientLock = new ReentrantLock();
  Lock serverLock = new ReentrantLock();
  ByteBuf buffer = Unpooled.buffer();
  Exception clientException = new Exception();
  Exception serverException = new Exception();
  EventLoopGroup eventLoopGroup =
      setUpServer(
          getServerInitializer(ssc.key(), ssc.cert(), serverLock, serverException), localAddress);
  SslClientInitializer<LocalChannel> sslClientInitializer =
      new SslClientInitializer<>(SslProvider.JDK, (X509Certificate[]) null);
  Channel channel =
      setUpClient(
          eventLoopGroup,
          getClientInitializer(sslClientInitializer, clientLock, buffer, clientException),
          localAddress,
          PROTOCOL);
  // Wait for handshake exception to throw.
  clientLock.lock();
  serverLock.lock();
  // The connection is now terminated, both the client side and the server side should get
  // exceptions (caught in the caughtException method in EchoHandler and DumpHandler,
  // respectively).
  assertThat(clientException).hasCauseThat().isInstanceOf(DecoderException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(SSLHandshakeException.class);
  assertThat(serverException).hasCauseThat().isInstanceOf(DecoderException.class);
  assertThat(serverException).hasCauseThat().hasCauseThat().isInstanceOf(SSLException.class);
  assertThat(channel.isActive()).isFalse();

  Future<?> unusedFuture = eventLoopGroup.shutdownGracefully().syncUninterruptibly();
}

@Test
public void testSuccess_customTrustManager_acceptCertSignedByTrustedCa() throws Exception {
  LocalAddress localAddress =
      new LocalAddress("CUSTOM_TRUST_MANAGER_ACCEPT_CERT_SIGNED_BY_TRUSTED_CA");
  Lock clientLock = new ReentrantLock();
  Lock serverLock = new ReentrantLock();
  ByteBuf buffer = Unpooled.buffer();
  Exception clientException = new Exception();
  Exception serverException = new Exception();

  // Generate a new key pair.
  KeyPair keyPair = getKeyPair();

  // Generate a self signed certificate, and use it to sign the key pair.
  SelfSignedCertificate ssc = new SelfSignedCertificate();
  X509Certificate cert = signKeyPair(ssc, keyPair, SSL_HOST);

  // Set up the server to use the signed cert and private key to perform handshake;
  PrivateKey privateKey = keyPair.getPrivate();
  EventLoopGroup eventLoopGroup =
      setUpServer(
          getServerInitializer(privateKey, cert, serverLock, serverException), localAddress);

  // Set up the client to trust the self signed cert used to sign the cert that server provides.
  SslClientInitializer<LocalChannel> sslClientInitializer =
      new SslClientInitializer<>(SslProvider.JDK, ssc.cert());
  Channel channel =
      setUpClient(
          eventLoopGroup,
          getClientInitializer(sslClientInitializer, clientLock, buffer, clientException),
          localAddress,
          PROTOCOL);

  verifySslChannel(channel, ImmutableList.of(cert), clientLock, serverLock, buffer, SSL_HOST);

  Future<?> unusedFuture = eventLoopGroup.shutdownGracefully().syncUninterruptibly();
}

@Test
public void testSuccess_trustAnyClientCert() throws Exception {
  SelfSignedCertificate serverSsc = new SelfSignedCertificate(SSL_HOST);
  LocalAddress localAddress = new LocalAddress("TRUST_ANY_CLIENT_CERT");
  Lock clientLock = new ReentrantLock();
  Lock serverLock = new ReentrantLock();
  ByteBuf buffer = Unpooled.buffer();
  Exception clientException = new Exception();
  Exception serverException = new Exception();
  EventLoopGroup eventLoopGroup =
      setUpServer(
          getServerInitializer(serverLock, serverException, serverSsc.key(), serverSsc.cert()),
          localAddress);
  SelfSignedCertificate clientSsc = new SelfSignedCertificate();
  Channel channel =
      setUpClient(
          eventLoopGroup,
          getClientInitializer(
              serverSsc.cert(),
              clientSsc.key(),
              clientSsc.cert(),
              clientLock,
              buffer,
              clientException),
          localAddress,
          PROTOCOL);

  SSLSession sslSession =
      verifySslChannel(
          channel, ImmutableList.of(serverSsc.cert()), clientLock, serverLock, buffer, SSL_HOST);
  // Verify that the SSL session gets the client cert. Note that this SslSession is for the client
  // channel, therefore its local certificates are the remote certificates of the SslSession for
  // the server channel, and vice versa.
  assertThat(sslSession.getLocalCertificates()).asList().containsExactly(clientSsc.cert());
  assertThat(sslSession.getPeerCertificates()).asList().containsExactly(serverSsc.cert());

  Future<?> unusedFuture = eventLoopGroup.shutdownGracefully().syncUninterruptibly();
}

@Test
public void testFailure_requireClientCertificate() throws Exception {
  SelfSignedCertificate serverSsc = new SelfSignedCertificate(SSL_HOST);
  LocalAddress localAddress = new LocalAddress("REQUIRE_CLIENT_CERT");
  Lock clientLock = new ReentrantLock();
  Lock serverLock = new ReentrantLock();
  ByteBuf buffer = Unpooled.buffer();
  Exception clientException = new Exception();
  Exception serverException = new Exception();
  EventLoopGroup eventLoopGroup =
      setUpServer(
          getServerInitializer(serverLock, serverException, serverSsc.key(), serverSsc.cert()),
          localAddress);
  Channel channel =
      setUpClient(
          eventLoopGroup,
          getClientInitializer(
              serverSsc.cert(),
              // No client cert/private key used.
              null,
              null,
              clientLock,
              buffer,
              clientException),
          localAddress,
          PROTOCOL);

  serverLock.lock();

  // When the server rejects the client during handshake due to lack of client certificate, only
  // the server throws an exception.
  assertThat(serverException).hasCauseThat().isInstanceOf(DecoderException.class);
  assertThat(serverException)
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(SSLHandshakeException.class);
  assertThat(channel.isActive()).isFalse();

  Future<?> unusedFuture = eventLoopGroup.shutdownGracefully().syncUninterruptibly();
}

/**
 * Sets up a client channel connecting to the give local address.
 *
 * @param eventLoopGroup the same {@link EventLoopGroup} that is used to bootstrap server.
 * @return the connected client channel.
 */
static Channel setUpClient(
    EventLoopGroup eventLoopGroup,
    ChannelInitializer<LocalChannel> clientInitializer,
    LocalAddress localAddress,
    BackendProtocol protocol)
    throws Exception {
  Bootstrap b =
      new Bootstrap()
          .group(eventLoopGroup)
          .channel(LocalChannel.class)
          .handler(clientInitializer)
          .attr(PROTOCOL_KEY, protocol);
  return b.connect(localAddress).syncUninterruptibly().channel();
}

@Test
public void testCancelBind() throws Exception {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    ChannelPromise promise = pipeline.channel().newPromise();
    assertTrue(promise.cancel(false));
    ChannelFuture future = pipeline.bind(new LocalAddress("test"), promise);
    assertTrue(future.isCancelled());
}

@Test
public void testCancelConnect() throws Exception {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    ChannelPromise promise = pipeline.channel().newPromise();
    assertTrue(promise.cancel(false));
    ChannelFuture future = pipeline.connect(new LocalAddress("test"), promise);
    assertTrue(future.isCancelled());
}

@Test
public void testCloseInFlush() throws Exception {

    LocalAddress addr = new LocalAddress("testCloseInFlush");

    ServerBootstrap sb = getLocalServerBootstrap();
    sb.bind(addr).sync().channel();

    Bootstrap cb = getLocalClientBootstrap();

    setInterest(Event.WRITE, Event.FLUSH, Event.CLOSE, Event.EXCEPTION);

    Channel clientChannel = cb.connect(addr).sync().channel();

    clientChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() {

        @Override
        public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            promise.addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    ctx.channel().close();
                }
            });
            super.write(ctx, msg, promise);
            ctx.channel().flush();
        }
    });

    clientChannel.write(createTestBuf(2000)).sync();
    clientChannel.closeFuture().sync();

    assertLog("WRITE\nFLUSH\nCLOSE\n");
}

@Test
public void testWritabilityChanged() throws Exception {

    LocalAddress addr = new LocalAddress("testWritabilityChanged");

    ServerBootstrap sb = getLocalServerBootstrap();
    sb.bind(addr).sync().channel();

    Bootstrap cb = getLocalClientBootstrap();

    setInterest(Event.WRITE, Event.FLUSH, Event.WRITABILITY);

    Channel clientChannel = cb.connect(addr).sync().channel();
    clientChannel.config().setWriteBufferLowWaterMark(512);
    clientChannel.config().setWriteBufferHighWaterMark(1024);

    ChannelFuture future = clientChannel.write(createTestBuf(2000));
    clientChannel.flush();
    future.sync();

    clientChannel.close().sync();

    assertLog(
        "WRITABILITY: writable=false\n" +
        "WRITABILITY: writable=true\n" +
        "WRITE\n" +
        "WRITABILITY: writable=false\n" +
        "FLUSH\n" +
        "WRITABILITY: writable=true\n");
}

@Test
public void testFlushInWritabilityChanged() throws Exception {

    LocalAddress addr = new LocalAddress("testFlushInWritabilityChanged");

    ServerBootstrap sb = getLocalServerBootstrap();
    sb.bind(addr).sync().channel();

    Bootstrap cb = getLocalClientBootstrap();

    setInterest(Event.WRITE, Event.FLUSH, Event.WRITABILITY);

    Channel clientChannel = cb.connect(addr).sync().channel();
    clientChannel.config().setWriteBufferLowWaterMark(512);
    clientChannel.config().setWriteBufferHighWaterMark(1024);

    clientChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
            if (!ctx.channel().isWritable()) {
                ctx.channel().flush();
            }
            ctx.fireChannelWritabilityChanged();
        }
    });

    assertTrue(clientChannel.isWritable());
    clientChannel.write(createTestBuf(2000)).sync();
    clientChannel.close().sync();

    assertLog(
        "WRITABILITY: writable=false\n" +
        "FLUSH\n" +
        "WRITABILITY: writable=true\n" +
        "WRITE\n" +
        "WRITABILITY: writable=false\n" +
        "FLUSH\n" +
        "WRITABILITY: writable=true\n");
}

static SocketAddress local(ChannelHandlerContext ctx) {
    Channel channel = ctx.channel();
    if (channel != null && channel.localAddress() != null) {
        return channel.localAddress();
    }
    return LocalAddress.ANY;
}

private void setupData() {
    sut.recordCommandLatency(LocalAddress.ANY, LocalAddress.ANY, CommandType.BGSAVE, MILLISECONDS.toNanos(100),
            MILLISECONDS.toNanos(1000));
    sut.recordCommandLatency(LocalAddress.ANY, LocalAddress.ANY, CommandType.BGSAVE, MILLISECONDS.toNanos(200),
            MILLISECONDS.toNanos(1000));
    sut.recordCommandLatency(LocalAddress.ANY, LocalAddress.ANY, CommandType.BGSAVE, MILLISECONDS.toNanos(300),
            MILLISECONDS.toNanos(1000));
}

@Test
public void testWritabilityChanged() throws Exception {

    LocalAddress addr = new LocalAddress("testWritabilityChanged");

    ServerBootstrap sb = getLocalServerBootstrap();
    sb.bind(addr).sync().channel();

    Bootstrap cb = getLocalClientBootstrap();

    setInterest(Event.WRITE, Event.FLUSH, Event.WRITABILITY);

    Channel clientChannel = cb.connect(addr).sync().channel();
    clientChannel.config().setWriteBufferLowWaterMark(512);
    clientChannel.config().setWriteBufferHighWaterMark(1024);

    ChannelFuture future = clientChannel.write(createTestBuf(2000));
    clientChannel.flush();
    future.sync();

    clientChannel.close().sync();

    assertLog(
        "WRITABILITY: writable=false\n" +
        "WRITABILITY: writable=true\n" +
        "WRITE\n" +
        "WRITABILITY: writable=false\n" +
        "FLUSH\n" +
        "WRITABILITY: writable=true\n");
}