DetectPeerCloseWithoutReadTest Class — netty Architecture

Source Code

transport-native-unix-common-tests/src/main/java/io/netty/channel/unix/tests/DetectPeerCloseWithoutReadTest.java lines 41–214

public abstract class DetectPeerCloseWithoutReadTest {
    protected abstract EventLoopGroup newGroup();
    protected abstract Class<? extends ServerChannel> serverChannel();
    protected abstract Class<? extends Channel> clientChannel();

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    public void clientCloseWithoutServerReadIsDetectedNoExtraReadRequested() throws InterruptedException {
        clientCloseWithoutServerReadIsDetected0(false);
    }

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    public void clientCloseWithoutServerReadIsDetectedExtraReadRequested() throws InterruptedException {
        clientCloseWithoutServerReadIsDetected0(true);
    }

    private void clientCloseWithoutServerReadIsDetected0(final boolean extraReadRequested)
            throws InterruptedException {
        EventLoopGroup serverGroup = null;
        EventLoopGroup clientGroup = null;
        Channel serverChannel = null;
        try {
            final CountDownLatch latch = new CountDownLatch(1);
            final AtomicInteger bytesRead = new AtomicInteger();
            final int expectedBytes = 100;
            serverGroup = newGroup();
            clientGroup = newGroup();
            ServerBootstrap sb = new ServerBootstrap();
            sb.group(serverGroup);
            sb.channel(serverChannel());
            // Ensure we read only one message per read() call and that we need multiple read()
            // calls to consume everything.
            sb.childOption(ChannelOption.AUTO_READ, false);
            sb.childOption(ChannelOption.MAX_MESSAGES_PER_READ, 1);
            sb.childOption(ChannelOption.RECVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(expectedBytes / 10));
            sb.childHandler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) {
                    ch.pipeline().addLast(new TestHandler(bytesRead, extraReadRequested, latch));
                }
            });

            serverChannel = sb.bind(new InetSocketAddress(0)).syncUninterruptibly().channel();

            Bootstrap cb = new Bootstrap();
            cb.group(serverGroup);
            cb.channel(clientChannel());
            cb.handler(new ChannelInboundHandlerAdapter());
            Channel clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
            ByteBuf buf = clientChannel.alloc().buffer(expectedBytes);
            buf.writerIndex(buf.writerIndex() + expectedBytes);
            clientChannel.writeAndFlush(buf).addListener(ChannelFutureListener.CLOSE);

            latch.await();
            assertEquals(expectedBytes, bytesRead.get());
        } finally {
            if (serverChannel != null) {
                serverChannel.close().syncUninterruptibly();
            }
            if (serverGroup != null) {
                serverGroup.shutdownGracefully();
            }
            if (clientGroup != null) {
                clientGroup.shutdownGracefully();
            }
        }
    }

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    public void serverCloseWithoutClientReadIsDetectedNoExtraReadRequested() throws InterruptedException {
        serverCloseWithoutClientReadIsDetected0(false);
    }

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    public void serverCloseWithoutClientReadIsDetectedExtraReadRequested() throws InterruptedException {
        serverCloseWithoutClientReadIsDetected0(true);
    }