testDatagram() — netty Function Reference

Source Code

codec-native-quic/src/test/java/io/netty/handler/codec/quic/QuicChannelDatagramTest.java lines 64–160

    private void testDatagram(Executor executor, boolean flushInReadComplete) throws Throwable {
        AtomicReference<QuicDatagramExtensionEvent> serverEventRef = new AtomicReference<>();

        QuicChannelValidationHandler serverHandler = new QuicChannelValidationHandler() {

            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                if (msg instanceof ByteBuf) {
                    final ChannelFuture future;
                    if (!flushInReadComplete) {
                        future = ctx.writeAndFlush(msg);
                    } else {
                        future = ctx.write(msg);
                    }
                    future.addListener(ChannelFutureListener.CLOSE);
                } else {
                    ctx.fireChannelRead(msg);
                }
            }

            @Override
            public void channelReadComplete(ChannelHandlerContext ctx) {
                if (flushInReadComplete) {
                    ctx.flush();
                }
            }

            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                if (evt instanceof QuicDatagramExtensionEvent) {
                    serverEventRef.set((QuicDatagramExtensionEvent) evt);
                }
                super.userEventTriggered(ctx, evt);
            }
        };
        Channel server = QuicTestUtils.newServer(QuicTestUtils.newQuicServerBuilder(executor)
                        .datagram(10, 10),
                InsecureQuicTokenHandler.INSTANCE, serverHandler , new ChannelInboundHandlerAdapter());
        InetSocketAddress address = (InetSocketAddress) server.localAddress();

        Promise<ByteBuf> receivedBuffer = ImmediateEventExecutor.INSTANCE.newPromise();
        AtomicReference<QuicDatagramExtensionEvent> clientEventRef = new AtomicReference<>();
        Channel channel = QuicTestUtils.newClient(QuicTestUtils.newQuicClientBuilder(executor)
                .datagram(10, 10));

        QuicChannelValidationHandler clientHandler = new QuicChannelValidationHandler() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                if (!receivedBuffer.trySuccess((ByteBuf) msg)) {
                    ReferenceCountUtil.release(msg);
                }
            }

            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                if (evt instanceof QuicDatagramExtensionEvent) {
                    clientEventRef.set((QuicDatagramExtensionEvent) evt);
                }
                super.userEventTriggered(ctx, evt);
            }

            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                receivedBuffer.tryFailure(cause);
                super.exceptionCaught(ctx, cause);
            }
        };

        try {
            QuicChannel quicChannel = QuicTestUtils.newQuicChannelBootstrap(channel)
                    .handler(clientHandler)
                    .remoteAddress(address)
                    .connect()
                    .get();
            quicChannel.writeAndFlush(Unpooled.copiedBuffer(data)).sync();

            ByteBuf buffer = receivedBuffer.get();
            ByteBuf expected = Unpooled.wrappedBuffer(data);
            assertEquals(expected, buffer);
            buffer.release();
            expected.release();