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

@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 ChannelHandlerAdapter() {
        @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");
}

@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 ChannelHandlerAdapter() {

        @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\n" +
        "FLUSH\n" +
        "CLOSE\n");
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  return NettyServerBuilder
      .forAddress(new LocalAddress("in-process-1"))
      .flowControlWindow(65 * 1024)
      .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .channelType(LocalServerChannel.class);
}

@Override
protected ManagedChannel createChannel() {
  NettyChannelBuilder builder = NettyChannelBuilder
      .forAddress(new LocalAddress("in-process-1"))
      .negotiationType(NegotiationType.PLAINTEXT)
      .channelType(LocalChannel.class)
      .flowControlWindow(65 * 1024)
      .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE);
  io.grpc.internal.TestingAccessor.setStatsImplementation(
      builder, createClientCensusStatsModule());
  return builder.build();
}

@Override
public void start(StartContext context) throws StartException {
    log.debug("connect client");
    try {
        this.clientInjector.getValue().connect(new LocalAddress("liveoak"));
    } catch (Exception e) {
        throw new StartException(e);
    }
}

@Before
public void setUp() throws Exception {
    ServerBootstrap b = new ServerBootstrap();

    LocalAddress address = LocalAddress.ANY;

    EventLoopGroup workerGroup = new NioEventLoopGroup();
    final FixAcceptorChannelInitializer<Channel> channelInitializer = new FixAcceptorChannelInitializer<>(
            workerGroup,
            new FixApplicationAdapter(),
            authenticator,
            new InMemorySessionRepository()
    );

    serverChannel = (LocalServerChannel) b.group(new NioEventLoopGroup())
            .channel(LocalServerChannel.class)
            .handler(channelInitializer)
            .childHandler(new FixApplicationAdapter())
            .validate()
            .bind(address)
            .sync()
            .channel();

    pipeline = serverChannel.pipeline();

    when(authenticator.authenticate(any(FixMessage.class))).thenReturn(true);
}

@Test
public void testFailure_customTrustManager_wrongHostnameInCertificate() throws Exception {
  LocalAddress localAddress = new LocalAddress("CUSTOM_TRUST_MANAGER_WRONG_HOSTNAME");
  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, "wrong.com");

  // 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);

  serverLock.lock();
  clientLock.lock();

  // When the client rejects the server cert due to wrong hostname, the client error is wrapped
  // several layers in the exception. The server also throws an exception.
  assertThat(clientException).hasCauseThat().isInstanceOf(DecoderException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(SSLHandshakeException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(SSLHandshakeException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(CertificateException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .hasMessageThat()
      .contains(SSL_HOST);
  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_CertSignedByOtherCA() throws Exception {
  // The self-signed cert of the CA.
  SelfSignedCertificate caSsc = new SelfSignedCertificate();
  KeyPair keyPair = getKeyPair();
  X509Certificate serverCert = signKeyPair(caSsc, keyPair, SSL_HOST);
  LocalAddress localAddress = new LocalAddress("CERT_SIGNED_BY_OTHER_CA");
  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,
              keyPair.getPrivate(),
              // Serving both the server cert, and the CA cert
              serverCert,
              caSsc.cert()),
          localAddress);
  SelfSignedCertificate clientSsc = new SelfSignedCertificate();
  Channel channel =
      setUpClient(
          eventLoopGroup,
          getClientInitializer(
              // Client trusts the CA cert
              caSsc.cert(),
              clientSsc.key(),
              clientSsc.cert(),
              clientLock,
              buffer,
              clientException),
          localAddress,
          PROTOCOL);

  SSLSession sslSession =
      verifySslChannel(
          channel,
          ImmutableList.of(serverCert, caSsc.cert()),
          clientLock,
          serverLock,
          buffer,
          SSL_HOST);

  assertThat(sslSession.getLocalCertificates()).asList().containsExactly(clientSsc.cert());
  assertThat(sslSession.getPeerCertificates())
      .asList()
      .containsExactly(serverCert, caSsc.cert())
      .inOrder();

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

@Test
public void testFailure_wrongHostnameInCertificate() throws Exception {
  SelfSignedCertificate serverSsc = new SelfSignedCertificate("wrong.com");
  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);
  SelfSignedCertificate clientSsc = new SelfSignedCertificate();
  Channel channel =
      setUpClient(
          eventLoopGroup,
          getClientInitializer(
              serverSsc.cert(),
              clientSsc.key(),
              clientSsc.cert(),
              clientLock,
              buffer,
              clientException),
          localAddress,
          PROTOCOL);

  serverLock.lock();
  clientLock.lock();

  // When the client rejects the server cert due to wrong hostname, the client error is wrapped
  // several layers in the exception. The server also throws an exception.
  assertThat(clientException).hasCauseThat().isInstanceOf(DecoderException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(SSLHandshakeException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(SSLHandshakeException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .isInstanceOf(CertificateException.class);
  assertThat(clientException)
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .hasCauseThat()
      .hasMessageThat()
      .contains(SSL_HOST);
  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 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);

    // What is supposed to happen from this point:
    //
    // 1. Because this write attempt has been made from a non-I/O thread,
    //    ChannelOutboundBuffer.pendingWriteBytes will be increased before
    //    write() event is really evaluated.
    //    -> channelWritabilityChanged() will be triggered,
    //       because the Channel became unwritable.
    //
    // 2. The write() event is handled by the pipeline in an I/O thread.
    //    -> write() will be triggered.
    //
    // 3. Once the write() event is handled, ChannelOutboundBuffer.pendingWriteBytes
    //    will be decreased.
    //    -> channelWritabilityChanged() will be triggered,
    //       because the Channel became writable again.
    //
    // 4. The message is added to the ChannelOutboundBuffer and thus
    //    pendingWriteBytes will be increased again.
    //    -> channelWritabilityChanged() will be triggered.
    //
    // 5. The flush() event causes the write request in theChannelOutboundBuffer
    //    to be removed.
    //    -> flush() and channelWritabilityChanged() will be triggered.
    //
    // Note that the channelWritabilityChanged() in the step 4 can occur between
    // the flush() and the channelWritabilityChanged() in the stap 5, because
    // the flush() is invoked from a non-I/O thread while the other are from
    // an I/O thread.

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

    clientChannel.flush();
    future.sync();

    clientChannel.close().sync();

    assertLog(
            // Case 1:
            "WRITABILITY: writable=false\n" +
            "WRITE\n" +
            "WRITABILITY: writable=false\n" +
            "WRITABILITY: writable=false\n" +
            "FLUSH\n" +
            "WRITABILITY: writable=true\n",
            // Case 2:
            "WRITABILITY: writable=false\n" +
            "WRITE\n" +
            "WRITABILITY: writable=false\n" +
            "FLUSH\n" +
            "WRITABILITY: writable=true\n" +
            "WRITABILITY: writable=true\n");
}