AbstractSocketReuseFdTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/AbstractSocketReuseFdTest.java lines 44–186

public abstract class AbstractSocketReuseFdTest extends AbstractSocketTest {
    @Override
    protected abstract SocketAddress newSocketAddress();

    @Override
    protected abstract List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories();

    @Test
    @Timeout(value = 60000, unit = TimeUnit.MILLISECONDS)
    public void testReuseFd(TestInfo testInfo) throws Throwable {
        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
            @Override
            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
                testReuseFd(serverBootstrap, bootstrap);
            }
        });
    }

    public void testReuseFd(ServerBootstrap sb, Bootstrap cb) throws Throwable {
        sb.childOption(ChannelOption.AUTO_READ, true);
        cb.option(ChannelOption.AUTO_READ, true);

        // Use a number which will typically not exceed /proc/sys/net/core/somaxconn (which is 128 on linux by default
        // often).
        int numChannels = 100;
        final AtomicReference<Throwable> globalException = new AtomicReference<Throwable>();
        final AtomicInteger serverRemaining = new AtomicInteger(numChannels);
        final AtomicInteger clientRemaining = new AtomicInteger(numChannels);
        final Promise<Void> serverDonePromise = ImmediateEventExecutor.INSTANCE.newPromise();
        final Promise<Void> clientDonePromise = ImmediateEventExecutor.INSTANCE.newPromise();

        sb.childHandler(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel sch) {
                ReuseFdHandler sh = new ReuseFdHandler(
                    false,
                    globalException,
                    serverRemaining,
                    serverDonePromise);
                sch.pipeline().addLast("handler", sh);
            }
        });

        cb.handler(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel sch) {
                ReuseFdHandler ch = new ReuseFdHandler(
                    true,
                    globalException,
                    clientRemaining,
                    clientDonePromise);
                sch.pipeline().addLast("handler", ch);
            }
        });

        ChannelFutureListener listener = future -> {
            if (!future.isSuccess()) {
                clientDonePromise.tryFailure(future.cause());
            }
        };

        Channel sc = sb.bind().sync().channel();
        for (int i = 0; i < numChannels; i++) {
            cb.connect(sc.localAddress()).addListener(listener);
        }

        clientDonePromise.sync();
        serverDonePromise.sync();
        sc.close().sync();

        if (globalException.get() != null && !(globalException.get() instanceof IOException)) {
            throw globalException.get();
        }
    }

    static class ReuseFdHandler extends ChannelInboundHandlerAdapter {
        private static final String EXPECTED_PAYLOAD = "payload";

        private final Promise<Void> donePromise;
        private final AtomicInteger remaining;
        private final boolean client;