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

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

public NetworkDispatcher connectToLocal(SocketAddress address)
{
    NetworkDispatcher dispatch = new NetworkDispatcher(this, NetworkSide.CLIENT);

    final EventLoopGroup boss = new DefaultEventLoopGroup();
    final Bootstrap b = new Bootstrap()
            .group(boss)
            .handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception
                {
                    ch.pipeline().addLast(dispatch);
                }
            })
            .channel(LocalChannel.class);

    //Connect and wait until done
    b.connect(address).syncUninterruptibly();

    return dispatch;
}

private ChannelInitializer<LocalChannel> getServerInitializer(
    PrivateKey privateKey,
    X509Certificate certificate,
    Lock serverLock,
    Exception serverException)
    throws Exception {
  SslContext sslContext = SslContextBuilder.forServer(privateKey, certificate).build();
  return new ChannelInitializer<LocalChannel>() {
    @Override
    protected void initChannel(LocalChannel ch) throws Exception {
      ch.pipeline()
          .addLast(
              sslContext.newHandler(ch.alloc()), new EchoHandler(serverLock, serverException));
    }
  };
}

private ChannelInitializer<LocalChannel> getServerInitializer(
    Lock serverLock,
    Exception serverException,
    PrivateKey privateKey,
    X509Certificate... certificates)
    throws Exception {
  return new ChannelInitializer<LocalChannel>() {
    @Override
    protected void initChannel(LocalChannel ch) throws Exception {
      ch.pipeline()
          .addLast(
              new SslServerInitializer<LocalChannel>(SslProvider.JDK, privateKey, certificates),
              new EchoHandler(serverLock, serverException));
    }
  };
}

/**
 * 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;
}

@Test
public void testRemoveChannelHandler() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    ChannelHandler handler1 = newHandler();
    ChannelHandler handler2 = newHandler();
    ChannelHandler handler3 = newHandler();

    pipeline.addLast("handler1", handler1);
    pipeline.addLast("handler2", handler2);
    pipeline.addLast("handler3", handler3);
    assertSame(pipeline.get("handler1"), handler1);
    assertSame(pipeline.get("handler2"), handler2);
    assertSame(pipeline.get("handler3"), handler3);

    pipeline.remove(handler1);
    assertNull(pipeline.get("handler1"));
    pipeline.remove(handler2);
    assertNull(pipeline.get("handler2"));
    pipeline.remove(handler3);
    assertNull(pipeline.get("handler3"));
}

@Test
public void testReplaceChannelHandler() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    ChannelHandler handler1 = newHandler();
    pipeline.addLast("handler1", handler1);
    pipeline.addLast("handler2", handler1);
    pipeline.addLast("handler3", handler1);
    assertSame(pipeline.get("handler1"), handler1);
    assertSame(pipeline.get("handler2"), handler1);
    assertSame(pipeline.get("handler3"), handler1);

    ChannelHandler newHandler1 = newHandler();
    pipeline.replace("handler1", "handler1", newHandler1);
    assertSame(pipeline.get("handler1"), newHandler1);

    ChannelHandler newHandler3 = newHandler();
    pipeline.replace("handler3", "handler3", newHandler3);
    assertSame(pipeline.get("handler3"), newHandler3);

    ChannelHandler newHandler2 = newHandler();
    pipeline.replace("handler2", "handler2", newHandler2);
    assertSame(pipeline.get("handler2"), newHandler2);
}

@Test
public void testFireChannelRegistered() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    ChannelPipeline pipeline = new LocalChannel().pipeline();
    pipeline.addLast(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                    latch.countDown();
                }
            });
        }
    });
    group.register(pipeline.channel());
    assertTrue(latch.await(2, TimeUnit.SECONDS));
}

@Before
public void setUp() throws Exception {

    when(ctx.channel()).thenReturn(new LocalChannel());

    MessageFactory messageFactory = ConfigParser.createDefault();
    message = messageFactory.newMessage(0x0200);

    pan = randomNumeric(19);
    cvv = randomAlphanumeric(3);
    track1 = randomAlphanumeric(10);
    track2 = randomAlphanumeric(20);
    track3 = randomAlphanumeric(30);

    message.setValue(2, pan, IsoType.NUMERIC, pan.length());
    message.setValue(112, cvv, IsoType.NUMERIC, 3);
    message.setValue(35, track2, IsoType.LLLVAR, 37);
    message.setValue(36, track3, IsoType.LLLVAR, 106);
    message.setValue(45, track1, IsoType.LLLVAR, 76);
}

@Test
public void testRemoveChannelHandler() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    ChannelHandler handler1 = newHandler();
    ChannelHandler handler2 = newHandler();
    ChannelHandler handler3 = newHandler();

    pipeline.addLast("handler1", handler1);
    pipeline.addLast("handler2", handler2);
    pipeline.addLast("handler3", handler3);
    assertSame(pipeline.get("handler1"), handler1);
    assertSame(pipeline.get("handler2"), handler2);
    assertSame(pipeline.get("handler3"), handler3);

    pipeline.remove(handler1);
    pipeline.remove(handler2);
    pipeline.remove(handler3);
}

@Test
public void testReplaceChannelHandler() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    ChannelHandler handler1 = newHandler();
    pipeline.addLast("handler1", handler1);
    pipeline.addLast("handler2", handler1);
    pipeline.addLast("handler3", handler1);
    assertSame(pipeline.get("handler1"), handler1);
    assertSame(pipeline.get("handler2"), handler1);
    assertSame(pipeline.get("handler3"), handler1);

    ChannelHandler newHandler1 = newHandler();
    pipeline.replace("handler1", "handler1", newHandler1);
    assertSame(pipeline.get("handler1"), newHandler1);

    ChannelHandler newHandler3 = newHandler();
    pipeline.replace("handler3", "handler3", newHandler3);
    assertSame(pipeline.get("handler3"), newHandler3);

    ChannelHandler newHandler2 = newHandler();
    pipeline.replace("handler2", "handler2", newHandler2);
    assertSame(pipeline.get("handler2"), newHandler2);
}

@Test
public void testFireChannelRegistered() throws Exception {
    ChannelPipeline pipeline = new LocalChannel().pipeline();
    group.register(pipeline.channel());
    final CountDownLatch latch = new CountDownLatch(1);
    pipeline.addLast(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                    latch.countDown();
                }
            });
        }
    });
    pipeline.fireChannelRegistered();
    assertTrue(latch.await(2, TimeUnit.SECONDS));
}

@Test
public void testRemoveChannelHandler() {
    ChannelPipeline pipeline = new LocalChannel(group.next()).pipeline();

    ChannelHandler handler1 = newHandler();
    ChannelHandler handler2 = newHandler();
    ChannelHandler handler3 = newHandler();

    pipeline.addLast("handler1", handler1);
    pipeline.addLast("handler2", handler2);
    pipeline.addLast("handler3", handler3);
    assertSame(pipeline.get("handler1"), handler1);
    assertSame(pipeline.get("handler2"), handler2);
    assertSame(pipeline.get("handler3"), handler3);

    pipeline.remove(handler1);
    pipeline.remove(handler2);
    pipeline.remove(handler3);
}

@Test
public void testReplaceChannelHandler() {
    ChannelPipeline pipeline = new LocalChannel(group.next()).pipeline();

    ChannelHandler handler1 = newHandler();
    pipeline.addLast("handler1", handler1);
    pipeline.addLast("handler2", handler1);
    pipeline.addLast("handler3", handler1);
    assertSame(pipeline.get("handler1"), handler1);
    assertSame(pipeline.get("handler2"), handler1);
    assertSame(pipeline.get("handler3"), handler1);

    ChannelHandler newHandler1 = newHandler();
    pipeline.replace("handler1", "handler1", newHandler1);
    assertSame(pipeline.get("handler1"), newHandler1);

    ChannelHandler newHandler3 = newHandler();
    pipeline.replace("handler3", "handler3", newHandler3);
    assertSame(pipeline.get("handler3"), newHandler3);

    ChannelHandler newHandler2 = newHandler();
    pipeline.replace("handler2", "handler2", newHandler2);
    assertSame(pipeline.get("handler2"), newHandler2);
}

@Test
public void testFireChannelRegistered() throws Exception {
    ChannelPipeline pipeline = new LocalChannel(group.next()).pipeline();
    final CountDownLatch latch = new CountDownLatch(1);
    pipeline.addLast(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new ChannelHandlerAdapter() {
                @Override
                public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                    latch.countDown();
                }
            });
        }
    });
    pipeline.fireChannelRegistered();
    assertTrue(latch.await(2, TimeUnit.SECONDS));
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
@SideOnly(Side.CLIENT)
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

private ChannelInitializer<LocalChannel> getClientInitializer(
    SslClientInitializer<LocalChannel> sslClientInitializer,
    Lock clientLock,
    ByteBuf buffer,
    Exception clientException) {
  return new ChannelInitializer<LocalChannel>() {
    @Override
    protected void initChannel(LocalChannel ch) throws Exception {
      ch.pipeline()
          .addLast(sslClientInitializer, new DumpHandler(clientLock, buffer, clientException));
    }
  };
}

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

private ChannelInitializer<LocalChannel> getClientInitializer(
    X509Certificate trustedCertificate,
    PrivateKey privateKey,
    X509Certificate certificate,
    Lock clientLock,
    ByteBuf buffer,
    Exception clientException) {
  return new ChannelInitializer<LocalChannel>() {
    @Override
    protected void initChannel(LocalChannel ch) throws Exception {
      SslContextBuilder sslContextBuilder =
          SslContextBuilder.forClient().trustManager(trustedCertificate);
      if (privateKey != null && certificate != null) {
        sslContextBuilder.keyManager(privateKey, certificate);
      }
      SslHandler sslHandler =
          sslContextBuilder.build().newHandler(ch.alloc(), SSL_HOST, SSL_PORT);

      // Enable hostname verification.
      SSLEngine sslEngine = sslHandler.engine();
      SSLParameters sslParameters = sslEngine.getSSLParameters();
      sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
      sslEngine.setSSLParameters(sslParameters);

      ch.pipeline().addLast("Client SSL Handler", sslHandler);
      ch.pipeline().addLast(new DumpHandler(clientLock, buffer, clientException));
    }
  };
}

/**
 * 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();
}

@Override
public Publisher<Long> createFailedPublisher() {
    LocalChannel channel = new LocalChannel();
    eventLoop.register(channel);
    HandlerPublisher<Long> publisher = new HandlerPublisher<>(channel.eventLoop(), Long.class);
    channel.pipeline().addLast("publisher", publisher);
    channel.pipeline().fireExceptionCaught(new RuntimeException("failed"));

    return publisher;
}

@Test
public void testChannelHandlerContextNavigation() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    final int HANDLER_ARRAY_LEN = 5;
    ChannelHandler[] firstHandlers = newHandlers(HANDLER_ARRAY_LEN);
    ChannelHandler[] lastHandlers = newHandlers(HANDLER_ARRAY_LEN);

    pipeline.addFirst(firstHandlers);
    pipeline.addLast(lastHandlers);

    verifyContextNumber(pipeline, HANDLER_ARRAY_LEN * 2);
}

@Test
public void testPipelineOperation() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    final int handlerNum = 5;
    ChannelHandler[] handlers1 = newHandlers(handlerNum);
    ChannelHandler[] handlers2 = newHandlers(handlerNum);

    final String prefixX = "x";
    for (int i = 0; i < handlerNum; i++) {
        if (i % 2 == 0) {
            pipeline.addFirst(prefixX + i, handlers1[i]);
        } else {
            pipeline.addLast(prefixX + i, handlers1[i]);
        }
    }

    for (int i = 0; i < handlerNum; i++) {
        if (i % 2 != 0) {
            pipeline.addBefore(prefixX + i, String.valueOf(i), handlers2[i]);
        } else {
            pipeline.addAfter(prefixX + i, String.valueOf(i), handlers2[i]);
        }
    }

    verifyContextNumber(pipeline, handlerNum * 2);
}

@Test
public void testChannelHandlerContextOrder() {
    ChannelPipeline pipeline = new LocalChannel().pipeline();

    pipeline.addFirst("1", newHandler());
    pipeline.addLast("10", newHandler());

    pipeline.addBefore("10", "5", newHandler());
    pipeline.addAfter("1", "3", newHandler());
    pipeline.addBefore("5", "4", newHandler());
    pipeline.addAfter("5", "6", newHandler());

    pipeline.addBefore("1", "0", newHandler());
    pipeline.addAfter("10", "11", newHandler());

    AbstractChannelHandlerContext ctx = (AbstractChannelHandlerContext) pipeline.firstContext();
    assertNotNull(ctx);
    while (ctx != null) {
        int i = toInt(ctx.name());
        int j = next(ctx);
        if (j != -1) {
            assertTrue(i < j);
        } else {
            assertNull(ctx.next.next);
        }
        ctx = ctx.next;
    }

    verifyContextNumber(pipeline, 8);
}

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