KQueueDomainSocketFdTest Class — netty Architecture

Source Code

transport-native-kqueue/src/test/java/io/netty/channel/kqueue/KQueueDomainSocketFdTest.java lines 41–111

public class KQueueDomainSocketFdTest extends AbstractSocketTest {
    @Override
    protected SocketAddress newSocketAddress() {
        return KQueueSocketTestPermutation.newSocketAddress();
    }

    @Override
    protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
        return KQueueSocketTestPermutation.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 {
        final BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>(1);
        sb.childHandler(new ChannelInboundHandlerAdapter() {
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                // Create new channel and obtain a file descriptor from it.
                final KQueueDomainSocketChannel ch = new KQueueDomainSocketChannel();

                ctx.writeAndFlush(ch.fd()).addListener(future -> {
                    if (!future.isSuccess()) {
                        Throwable cause = future.cause();
                        queue.offer(cause);
                    }
                });
            }
        });
        cb.handler(new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                FileDescriptor fd = (FileDescriptor) msg;
                queue.offer(fd);
            }

            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                queue.add(cause);
                ctx.close();
            }
        });
        cb.option(KQueueChannelOption.DOMAIN_SOCKET_READ_MODE,
                DomainSocketReadMode.FILE_DESCRIPTORS);
        Channel sc = sb.bind().sync().channel();
        Channel cc = cb.connect(sc.localAddress()).sync().channel();

        Object received = queue.take();
        cc.close().sync();
        sc.close().sync();

        if (received instanceof FileDescriptor) {
            FileDescriptor fd = (FileDescriptor) received;
            assertTrue(fd.isOpen());
            fd.close();
            assertFalse(fd.isOpen());
            assertNull(queue.poll());
        } else {
            throw (Throwable) received;
        }
    }
}