io.netty.channel.embedded.EmbeddedChannel#pipeline ( )源码实例Demo

@Test
public void testSuccess_swappedInitializerWithSslHandler() throws Exception {
  SelfSignedCaCertificate ssc = SelfSignedCaCertificate.create(SSL_HOST);
  SslServerInitializer<EmbeddedChannel> sslServerInitializer =
      new SslServerInitializer<>(
          true,
          false,
          sslProvider,
          Suppliers.ofInstance(ssc.key()),
          Suppliers.ofInstance(ImmutableList.of(ssc.cert())));
  EmbeddedChannel channel = new EmbeddedChannel();
  ChannelPipeline pipeline = channel.pipeline();
  pipeline.addLast(sslServerInitializer);
  ChannelHandler firstHandler = pipeline.first();
  assertThat(firstHandler.getClass()).isEqualTo(SslHandler.class);
  SslHandler sslHandler = (SslHandler) firstHandler;
  assertThat(sslHandler.engine().getNeedClientAuth()).isTrue();
  assertThat(channel.isActive()).isTrue();
}
 

@Test
public void testSuccess_swappedInitializerWithSslHandler() throws Exception {
  SslClientInitializer<EmbeddedChannel> sslClientInitializer =
      new SslClientInitializer<>(
          sslProvider, hostProvider, portProvider, ImmutableList.of(), null, null);
  EmbeddedChannel channel = new EmbeddedChannel();
  ChannelPipeline pipeline = channel.pipeline();
  pipeline.addLast(sslClientInitializer);
  ChannelHandler firstHandler = pipeline.first();
  assertThat(firstHandler.getClass()).isEqualTo(SslHandler.class);
  SslHandler sslHandler = (SslHandler) firstHandler;
  assertThat(sslHandler.engine().getPeerHost()).isEqualTo(SSL_HOST);
  assertThat(sslHandler.engine().getPeerPort()).isEqualTo(SSL_PORT);
  assertThat(channel.isActive()).isTrue();
}
 

@Test
public void serverSdsHandler_defaultDownstreamTlsContext_expectFallbackProtocolNegotiator()
    throws IOException {
  ChannelHandler mockChannelHandler = mock(ChannelHandler.class);
  ProtocolNegotiator mockProtocolNegotiator = mock(ProtocolNegotiator.class);
  when(mockProtocolNegotiator.newHandler(grpcHandler)).thenReturn(mockChannelHandler);
  // we need InetSocketAddress instead of EmbeddedSocketAddress as localAddress for this test
  channel =
      new EmbeddedChannel() {
        @Override
        public SocketAddress localAddress() {
          return new InetSocketAddress("172.168.1.1", 80);
        }
      };
  pipeline = channel.pipeline();
  DownstreamTlsContext downstreamTlsContext =
      DownstreamTlsContext.fromEnvoyProtoDownstreamTlsContext(
          io.envoyproxy.envoy.api.v2.auth.DownstreamTlsContext.getDefaultInstance());

  XdsClientWrapperForServerSds xdsClientWrapperForServerSds =
      XdsClientWrapperForServerSdsTest.createXdsClientWrapperForServerSds(
          80, downstreamTlsContext);
  SdsProtocolNegotiators.HandlerPickerHandler handlerPickerHandler =
      new SdsProtocolNegotiators.HandlerPickerHandler(
          grpcHandler, xdsClientWrapperForServerSds, mockProtocolNegotiator);
  pipeline.addLast(handlerPickerHandler);
  channelHandlerCtx = pipeline.context(handlerPickerHandler);
  assertThat(channelHandlerCtx).isNotNull(); // should find HandlerPickerHandler

  // kick off protocol negotiation: should replace HandlerPickerHandler with ServerSdsHandler
  pipeline.fireUserEventTriggered(InternalProtocolNegotiationEvent.getDefault());
  channelHandlerCtx = pipeline.context(handlerPickerHandler);
  assertThat(channelHandlerCtx).isNull();
  channel.runPendingTasks(); // need this for tasks to execute on eventLoop
  Iterator<Map.Entry<String, ChannelHandler>> iterator = pipeline.iterator();
  assertThat(iterator.next().getValue()).isSameInstanceAs(mockChannelHandler);
  // no more handlers in the pipeline
  assertThat(iterator.hasNext()).isFalse();
}
 

@Test
public void testSslRejection() {
    PostgresWireProtocol ctx =
        new PostgresWireProtocol(
            mock(SQLOperations.class),
            sessionContext -> AccessControl.DISABLED,
            new AlwaysOKNullAuthentication(),
            null);

    channel = new EmbeddedChannel(ctx.decoder, ctx.handler);

    ByteBuf buffer = Unpooled.buffer();
    ClientMessages.sendSslReqMessage(buffer);
    channel.writeInbound(buffer);

    // We should get back an 'N'...
    ByteBuf responseBuffer = channel.readOutbound();
    try {
        byte response = responseBuffer.readByte();
        assertEquals(response, 'N');
    } finally {
        responseBuffer.release();
    }

    // ...and continue unencrypted (no ssl handler)
    for (Map.Entry<String, ChannelHandler> entry : channel.pipeline()) {
        assertThat(entry.getValue(), isOneOf(ctx.decoder, ctx.handler));
    }
}
 

@Test
public void serverSdsHandler_addLast() throws IOException {
  // we need InetSocketAddress instead of EmbeddedSocketAddress as localAddress for this test
  channel =
      new EmbeddedChannel() {
        @Override
        public SocketAddress localAddress() {
          return new InetSocketAddress("172.168.1.1", 80);
        }
      };
  pipeline = channel.pipeline();
  DownstreamTlsContext downstreamTlsContext =
      buildDownstreamTlsContextFromFilenames(SERVER_1_KEY_FILE, SERVER_1_PEM_FILE, CA_PEM_FILE);

  XdsClientWrapperForServerSds xdsClientWrapperForServerSds =
      XdsClientWrapperForServerSdsTest.createXdsClientWrapperForServerSds(
          80, downstreamTlsContext);
  SdsProtocolNegotiators.HandlerPickerHandler handlerPickerHandler =
      new SdsProtocolNegotiators.HandlerPickerHandler(grpcHandler, xdsClientWrapperForServerSds,
          InternalProtocolNegotiators.serverPlaintext());
  pipeline.addLast(handlerPickerHandler);
  channelHandlerCtx = pipeline.context(handlerPickerHandler);
  assertThat(channelHandlerCtx).isNotNull(); // should find HandlerPickerHandler

  // kick off protocol negotiation: should replace HandlerPickerHandler with ServerSdsHandler
  pipeline.fireUserEventTriggered(InternalProtocolNegotiationEvent.getDefault());
  channelHandlerCtx = pipeline.context(handlerPickerHandler);
  assertThat(channelHandlerCtx).isNull();
  channelHandlerCtx = pipeline.context(SdsProtocolNegotiators.ServerSdsHandler.class);
  assertThat(channelHandlerCtx).isNotNull();
  channel.runPendingTasks(); // need this for tasks to execute on eventLoop
  channelHandlerCtx = pipeline.context(SdsProtocolNegotiators.ServerSdsHandler.class);
  assertThat(channelHandlerCtx).isNull();

  // pipeline should only have SslHandler and ServerTlsHandler
  Iterator<Map.Entry<String, ChannelHandler>> iterator = pipeline.iterator();
  assertThat(iterator.next().getValue()).isInstanceOf(SslHandler.class);
  // ProtocolNegotiators.ServerTlsHandler.class is not accessible, get canonical name
  assertThat(iterator.next().getValue().getClass().getCanonicalName())
      .contains("ProtocolNegotiators.ServerTlsHandler");
}