testStatsAreCollected() — netty Function Reference

Source Code

codec-native-quic/src/test/java/io/netty/handler/codec/quic/QuicConnectionStatsTest.java lines 38–131

    @ParameterizedTest
    @MethodSource("newSslTaskExecutors")
    public void testStatsAreCollected(Executor executor) throws Throwable {
        Channel server = null;
        Channel channel = null;
        AtomicInteger counter = new AtomicInteger();

        Promise<QuicConnectionStats> serverActiveStats = ImmediateEventExecutor.INSTANCE.newPromise();
        Promise<QuicConnectionStats> serverInactiveStats = ImmediateEventExecutor.INSTANCE.newPromise();
        QuicChannelValidationHandler serverHandler = new QuicChannelValidationHandler() {
            @Override
            public void channelActive(ChannelHandlerContext ctx) {
                super.channelActive(ctx);
                collectStats(ctx, serverActiveStats);
            }

            @Override
            public void channelInactive(ChannelHandlerContext ctx) {
                collectStats(ctx, serverInactiveStats);
                ctx.fireChannelInactive();
            }

            private void collectStats(ChannelHandlerContext ctx, Promise<QuicConnectionStats> promise) {
                QuicheQuicChannel channel = (QuicheQuicChannel) ctx.channel();
                channel.collectStats(promise);
            }
        };
        QuicChannelValidationHandler clientHandler = new QuicChannelValidationHandler();
        try {
            server = QuicTestUtils.newServer(executor, serverHandler, new ChannelInboundHandlerAdapter() {

                @Override
                public void channelActive(ChannelHandlerContext ctx) {
                    counter.incrementAndGet();
                }

                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    // Let's just echo back the message.
                    ctx.writeAndFlush(msg);
                }

                @Override
                public boolean isSharable() {
                    return true;
                }
            });
            channel = QuicTestUtils.newClient(executor);

            QuicChannel quicChannel = QuicTestUtils.newQuicChannelBootstrap(channel)
                    .handler(clientHandler)
                    .streamHandler(new ChannelInboundHandlerAdapter())
                    .remoteAddress(server.localAddress())
                    .connect().get();
            assertNotNull(quicChannel.collectStats().sync().getNow());
            quicChannel.createStream(QuicStreamType.BIDIRECTIONAL, new ChannelInboundHandlerAdapter() {
                private final int bufferSize = 8;
                private int received;

                @Override
                public void channelActive(ChannelHandlerContext ctx) {
                    ctx.writeAndFlush(Unpooled.buffer().writeZero(bufferSize));
                }

                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ByteBuf buffer = (ByteBuf) msg;
                    received += buffer.readableBytes();
                    buffer.release();
                    if (received == bufferSize) {
                        ctx.close().addListener((ChannelFuture future) -> {
                            // Close the underlying QuicChannel as well.
                            future.channel().parent().close();
                        });
                    }
                }
            }).sync();

            // Wait until closure
            quicChannel.closeFuture().sync();
            assertStats(quicChannel.collectStats().sync().getNow());