IoUringDomainSocketFdTest Class — netty Architecture

Source Code

transport-native-io_uring/src/test/java/io/netty/channel/uring/IoUringDomainSocketFdTest.java lines 45–156

public class IoUringDomainSocketFdTest extends AbstractSocketTest {
    @Override
    protected SocketAddress newSocketAddress() {
        return IoUringSocketTestPermutation.newDomainSocketAddress();
    }

    @Override
    protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
        return IoUringSocketTestPermutation.INSTANCE.domainSocket();
    }

    @Test
    @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
    public void testSendRecvFd(TestInfo testInfo) throws Throwable {
        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
            @Override
            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
                testSendRecvFd(serverBootstrap, bootstrap);
            }
        });
    }

    public void testSendRecvFd(ServerBootstrap sb, Bootstrap cb) throws Throwable {

        String expected = "Hello World";
        CompletableFuture<FileDescriptor> recvFdFuture = new CompletableFuture<>();
        CompletableFuture<ByteBuf> recvByteBufFuture = new CompletableFuture<>();

        sb.childHandler(new ChannelInboundHandlerAdapter() {
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                // Create new channel and obtain a file descriptor from it.
                final IoUringDomainSocketChannel ch = new IoUringDomainSocketChannel();

                ctx.writeAndFlush(ch.fd()).addListener(future -> {
                    if (!future.isSuccess()) {
                        Throwable cause = future.cause();
                        recvFdFuture.completeExceptionally(cause);
                    } else {
                        ByteBuf sendBuffer = ctx.alloc().directBuffer(expected.length());
                        sendBuffer.writeBytes(expected.getBytes());
                        ctx.writeAndFlush(sendBuffer).addListener(f -> {
                            if (!f.isSuccess()) {
                                Throwable cause = f.cause();
                                recvByteBufFuture.completeExceptionally(cause);
                            }
                        });
                    }
                });
            }
        });

        cb.handler(new ChannelInboundHandlerAdapter() {

            private CompositeByteBuf byteBufs;

            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                if (msg instanceof FileDescriptor) {
                    FileDescriptor fd = (FileDescriptor) msg;
                    recvFdFuture.complete(fd);
                    ctx.channel().config()
                            .setOption(IoUringChannelOption.DOMAIN_SOCKET_READ_MODE, DomainSocketReadMode.BYTES);
                } else {
                    byteBufs.addComponent(true, (ByteBuf) msg);
                }
            }

            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                DomainSocketReadMode readMode = ctx.channel().config()
                        .getOption(IoUringChannelOption.DOMAIN_SOCKET_READ_MODE);
                if (readMode == DomainSocketReadMode.FILE_DESCRIPTORS) {
                    recvFdFuture.completeExceptionally(cause);
                } else {
                    recvByteBufFuture.completeExceptionally(cause);
                }
                ctx.close();
            }

            @Override