testStagedExecution() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java lines 93–235

    @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();
            }
            if (h2.exception.get() != null) {
                throw h2.exception.get();
            }
            if (h3.exception.get() != null) {
                throw h3.exception.get();
            }

            Thread.sleep(10);
        }

        String currentName = Thread.currentThread().getName();

        try {
            // Events should never be handled from the current thread.
            assertFalse(h1.inboundThreadNames.contains(currentName));
            assertFalse(h2.inboundThreadNames.contains(currentName));
            assertFalse(h3.inboundThreadNames.contains(currentName));
            assertFalse(h1.outboundThreadNames.contains(currentName));
            assertFalse(h2.outboundThreadNames.contains(currentName));
            assertFalse(h3.outboundThreadNames.contains(currentName));
            assertFalse(h1.removalThreadNames.contains(currentName));
            assertFalse(h2.removalThreadNames.contains(currentName));
            assertFalse(h3.removalThreadNames.contains(currentName));

            // Assert that events were handled by the correct executor.
            for (String name: h1.inboundThreadNames) {
                assertTrue(name.startsWith("l-"));
            }
            for (String name: h2.inboundThreadNames) {
                assertTrue(name.startsWith("e1-"));
            }
            for (String name: h3.inboundThreadNames) {
                assertTrue(name.startsWith("e2-"));
            }
            for (String name: h1.outboundThreadNames) {
                assertTrue(name.startsWith("l-"));
            }
            for (String name: h2.outboundThreadNames) {
                assertTrue(name.startsWith("e1-"));
            }
            for (String name: h3.outboundThreadNames) {