testShutdownInputClosureByServerCausesStreamStopped() — netty Function Reference

Source Code

codec-native-quic/src/test/java/io/netty/handler/codec/quic/QuicStreamShutdownTest.java lines 87–153

    @ParameterizedTest
    @MethodSource("newSslTaskExecutors")
    public void testShutdownInputClosureByServerCausesStreamStopped(Executor executor) throws Throwable {
        Channel server = null;
        Channel channel = null;
        CountDownLatch latch = new CountDownLatch(2);
        AtomicReference<Throwable> errorRef = new AtomicReference<>();
        try {
            server = QuicTestUtils.newServer(executor, new ChannelInboundHandlerAdapter(),
                    new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            QuicStreamChannel streamChannel = (QuicStreamChannel) ctx.channel();
                            streamChannel.shutdownInput()
                                    .addListener(f -> {
                                        ByteBuf buffer = (ByteBuf) msg;
                                        if (!f.isSuccess()) {
                                            errorRef.compareAndSet(null, f.cause());
                                            latch.countDown();
                                            buffer.release();
                                        } else {
                                            ctx.writeAndFlush(buffer).addListener(channelFuture -> {
                                                if (!channelFuture.isSuccess()) {
                                                    errorRef.compareAndSet(null, channelFuture.cause());
                                                }
                                                latch.countDown();
                                            });
                                        }
                                    });
                        }
                    });

            channel = QuicTestUtils.newClient(executor);
            QuicChannel quicChannel = QuicTestUtils.newQuicChannelBootstrap(channel)
                    .handler(new ChannelInboundHandlerAdapter())
                    .streamHandler(new ChannelInboundHandlerAdapter())
                    .remoteAddress(server.localAddress())
                    .connect()
                    .get();

            QuicStreamChannel streamChannel = quicChannel.createStream(QuicStreamType.BIDIRECTIONAL,
                    new ByteToMessageDecoder() {
                        @Override
                        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> list) {
                            if (in.readableBytes() == 4) {
                                latch.countDown();
                                ctx.close();
                            }
                        }
                    }).sync().getNow();

            streamChannel.writeAndFlush(Unpooled.buffer().writeInt(4)).sync();

            if (!latch.await(5, TimeUnit.SECONDS)) {
                fail("Timeout while waiting for completion", errorRef.get());
            }
            Throwable error = errorRef.get();
            if (error != null) {
                fail("Failure during execution", error);
            }
        } finally {
            QuicTestUtils.closeIfNotNull(channel);
            QuicTestUtils.closeIfNotNull(server);

            shutdown(executor);
        }
    }