testShutdownOutputClosureByServerCausesStreamReset() — netty Function Reference

Source Code

codec-native-quic/src/test/java/io/netty/handler/codec/quic/QuicStreamShutdownTest.java lines 155–219

    @ParameterizedTest
    @MethodSource("newSslTaskExecutors")
    public void testShutdownOutputClosureByServerCausesStreamReset(Executor executor) throws Throwable {
        Channel server = null;
        Channel channel = null;
        CountDownLatch latch = new CountDownLatch(2);
        AtomicReference<Throwable> errorRef = new AtomicReference<>();
        AtomicLong applicationErrorCode = new AtomicLong(-1);
        try {
            server = QuicTestUtils.newServer(executor, new ChannelInboundHandlerAdapter(),
                    new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            QuicStreamChannel streamChannel = (QuicStreamChannel) ctx.channel();
                            ByteBuf buffer = (ByteBuf) msg;
                            buffer.release();
                            streamChannel.shutdownOutput(100)
                                    .addListener(f -> {
                                        if (!f.isSuccess()) {
                                            errorRef.compareAndSet(null, f.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 ChannelInboundHandlerAdapter() {
                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                            if (cause instanceof QuicStreamResetException) {
                                QuicStreamResetException reset = (QuicStreamResetException) cause;
                                applicationErrorCode.set(reset.applicationProtocolCode());
                            } else {
                                errorRef.set(cause);
                            }
                            latch.countDown();
                        }
                    }).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);
            }
            assertEquals(100, applicationErrorCode.get());
        } finally {
            QuicTestUtils.closeIfNotNull(channel);
            QuicTestUtils.closeIfNotNull(server);

            shutdown(executor);
        }
    }