testSegmentedDatagramPacket() — netty Function Reference

Source Code

transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollDatagramUnicastTest.java lines 114–193

    private void testSegmentedDatagramPacket(Bootstrap sb, Bootstrap cb, boolean composite, boolean gro)
            throws Throwable {
        if (!(cb.group() instanceof IoEventLoopGroup &&
                ((IoEventLoopGroup) cb.group()).isIoType(EpollIoHandler.class))) {
            // Only supported for the native epoll transport.
            return;
        }

        if (gro && !(sb.group() instanceof IoEventLoopGroup) &&
                ((IoEventLoopGroup) sb.group()).isIoType(EpollIoHandler.class)) {
            // Only supported for the native epoll transport.
            return;
        }
        assumeTrue(EpollDatagramChannel.isSegmentedDatagramPacketSupported());
        Channel sc = null;
        Channel cc = null;

        try {
            cb.handler(new SimpleChannelInboundHandler<Object>() {
                @Override
                public void channelRead0(ChannelHandlerContext ctx, Object msgs) {
                    // Nothing will be sent.
                }
            });

            cc = cb.bind(newSocketAddress()).sync().channel();

            final int numBuffers = 16;
            final int segmentSize = 512;
            int bufferCapacity = numBuffers * segmentSize;
            final CountDownLatch latch = new CountDownLatch(numBuffers);
            AtomicReference<Throwable> errorRef = new AtomicReference<Throwable>();
            if (gro) {
                // Enable GRO and also ensure we can read everything with one read as otherwise
                // we will drop things on the floor.
                sb.option(EpollChannelOption.UDP_GRO, true);
                sb.option(ChannelOption.RECVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(bufferCapacity));
            }
            sc = sb.handler(new SimpleChannelInboundHandler<DatagramPacket>() {
                @Override
                public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) {
                    if (packet.content().readableBytes() == segmentSize) {
                        latch.countDown();
                    }
                }
            }).bind(newSocketAddress()).sync().channel();

            if (sc instanceof EpollDatagramChannel) {
                assertEquals(gro, sc.config().getOption(EpollChannelOption.UDP_GRO));
            }
            InetSocketAddress addr = sendToAddress((InetSocketAddress) sc.localAddress());
            final ByteBuf buffer;
            if (composite) {
                CompositeByteBuf compositeBuffer = Unpooled.compositeBuffer();
                for (int i = 0; i < numBuffers; i++) {
                    compositeBuffer.addComponent(true,
                            Unpooled.directBuffer(segmentSize).writeZero(segmentSize));
                }
                buffer = compositeBuffer;
            } else {
                buffer = Unpooled.directBuffer(bufferCapacity).writeZero(bufferCapacity);
            }
            cc.writeAndFlush(new io.netty.channel.unix.SegmentedDatagramPacket(buffer, segmentSize, addr)).sync();

            if (!latch.await(10, TimeUnit.SECONDS)) {
                Throwable error = errorRef.get();
                if (error != null) {
                    throw error;
                }
                fail();
            }
        } finally {
            if (cc != null) {
                cc.close().sync();
            }
            if (sc != null) {
                sc.close().sync();
            }
        }
    }