LocalTransportThreadModelTest Class — netty Architecture

Source Code

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java lines 51–612

public class LocalTransportThreadModelTest {

    private static EventLoopGroup group;
    private static LocalAddress localAddr;

    @BeforeAll
    public static void init() {
        // Configure a test server
        group = new MultiThreadIoEventLoopGroup(LocalIoHandler.newFactory());
        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 ChannelInboundHandlerAdapter() {
                      @Override
                      public void channelRead(ChannelHandlerContext ctx, Object msg) {
                          // Discard
                          ReferenceCountUtil.release(msg);
                      }
                  });
              }
          });

        localAddr = (LocalAddress) sb.bind(LocalAddress.ANY).syncUninterruptibly().channel().localAddress();
    }

    @AfterAll
    public static void destroy() throws Exception {
        group.shutdownGracefully().sync();
    }

    @Test
    @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
    @Disabled("regression test")
    public void testStagedExecutionMultiple() throws Throwable {
        for (int i = 0; i < 10; i ++) {
            testStagedExecution();
        }
    }

    @Test
    @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
    public void testStagedExecution() throws Throwable {
        EventLoopGroup l = new MultiThreadIoEventLoopGroup(
                4, new DefaultThreadFactory("l"), LocalIoHandler.newFactory());
        EventExecutorGroup e1 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e1"));
        EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
        ThreadNameAuditor h1 = new ThreadNameAuditor();
        ThreadNameAuditor h2 = new ThreadNameAuditor();
        ThreadNameAuditor h3 = new ThreadNameAuditor(true);

        Channel ch = new LocalChannel();
        // With no EventExecutor specified, h1 will be always invoked by EventLoop 'l'.
        ch.pipeline().addLast(h1);
        // h2 will be always invoked by EventExecutor 'e1'.
        ch.pipeline().addLast(e1, h2);
        // h3 will be always invoked by EventExecutor 'e2'.
        ch.pipeline().addLast(e2, h3);

        l.register(ch).sync().channel().connect(localAddr).sync();

        // Fire inbound events from all possible starting points.
        ch.pipeline().fireChannelRead("1");
        ch.pipeline().context(h1).fireChannelRead("2");
        ch.pipeline().context(h2).fireChannelRead("3");
        ch.pipeline().context(h3).fireChannelRead("4");
        // Fire outbound events from all possible starting points.
        ch.pipeline().write("5");
        ch.pipeline().context(h3).write("6");
        ch.pipeline().context(h2).write("7");
        ch.pipeline().context(h1).writeAndFlush("8").sync();

        ch.close().sync();

        // Wait until all events are handled completely.
        while (h1.outboundThreadNames.size() < 3 || h3.inboundThreadNames.size() < 3 ||
               h1.removalThreadNames.size() < 1) {
            if (h1.exception.get() != null) {
                throw h1.exception.get();