testBytesUntilUnwritable() — netty Function Reference

Source Code

codec-native-quic/src/test/java/io/netty/handler/codec/quic/QuicWritableTest.java lines 165–293

    @ParameterizedTest
    @MethodSource("newSslTaskExecutors")
    @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
    public void testBytesUntilUnwritable(Executor executor) throws Throwable  {
        Promise<Void> writePromise = ImmediateEventExecutor.INSTANCE.newPromise();
        final AtomicReference<Throwable> serverErrorRef = new AtomicReference<>();
        final AtomicReference<Throwable> clientErrorRef = new AtomicReference<>();
        final CountDownLatch writableAgainLatch = new CountDownLatch(1);
        int firstWriteNumBytes = 8;
        int maxData = 32 * 1024;
        final AtomicLong beforeWritableRef = new AtomicLong();
        QuicChannelValidationHandler serverHandler = new QuicChannelValidationHandler();
        Channel server = QuicTestUtils.newServer(
                QuicTestUtils.newQuicServerBuilder(executor).initialMaxStreamsBidirectional(5000),
                InsecureQuicTokenHandler.INSTANCE,
                serverHandler, new ChannelInboundHandlerAdapter() {

                    private int numBytesRead;
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) {
                        ByteBuf buffer = (ByteBuf) msg;
                        numBytesRead += buffer.readableBytes();
                        buffer.release();
                        if (numBytesRead == firstWriteNumBytes) {
                            long before = ctx.channel().bytesBeforeUnwritable();
                            beforeWritableRef.set(before);
                            assertTrue(before > 0);

                            while (before != 0) {
                                int size = (int) Math.min(before, 1024);
                                ctx.write(ctx.alloc().buffer(size).writeZero(size));
                                long newBefore = ctx.channel().bytesBeforeUnwritable();

                                assertEquals(before, newBefore + size);
                                before = newBefore;
                            }
                            ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(new PromiseNotifier<>(writePromise));
                        }
                    }

                    @Override
                    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
                        if (ctx.channel().isWritable()) {
                            if (ctx.channel().bytesBeforeUnwritable() > 0) {
                                writableAgainLatch.countDown();
                            }
                        }
                    }

                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                        serverErrorRef.set(cause);
                    }

                    @Override
                    public boolean isSharable() {
                        return true;
                    }
                });
        InetSocketAddress address = (InetSocketAddress) server.localAddress();
        Channel channel = QuicTestUtils.newClient(QuicTestUtils.newQuicClientBuilder(executor)
                .initialMaxStreamDataBidirectionalLocal(maxData));

        QuicChannelValidationHandler clientHandler = new QuicChannelValidationHandler();
        try {
            QuicChannel quicChannel = QuicTestUtils.newQuicChannelBootstrap(channel)
                    .handler(clientHandler)
                    .streamHandler(new ChannelInboundHandlerAdapter())
                    .remoteAddress(address)
                    .connect()
                    .get();
            QuicStreamChannel stream = quicChannel.createStream(
                    QuicStreamType.BIDIRECTIONAL, new ChannelInboundHandlerAdapter() {
                        int bytes;

                        @Override
                        public void channelRegistered(ChannelHandlerContext ctx) {
                            ctx.channel().config().setAutoRead(false);
                        }

                        @Override