EpollSpliceTest Class — netty Architecture

Source Code

transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSpliceTest.java lines 46–312

public class EpollSpliceTest {

    private static final int SPLICE_LEN = 32 * 1024;
    private static final Random random = new Random();
    private static final byte[] data = new byte[1048576];

    static {
        random.nextBytes(data);
    }

    @Test
    public void spliceToSocket() throws Throwable {
        final EchoHandler sh = new EchoHandler();
        final EchoHandler ch = new EchoHandler();

        EventLoopGroup group = new MultiThreadIoEventLoopGroup(1, EpollIoHandler.newFactory());
        ServerBootstrap bs = new ServerBootstrap();
        bs.channel(EpollServerSocketChannel.class);
        bs.group(group).childHandler(sh);
        final Channel sc = bs.bind(NetUtil.LOCALHOST, 0).syncUninterruptibly().channel();

        ServerBootstrap bs2 = new ServerBootstrap();
        bs2.channel(EpollServerSocketChannel.class);
        bs2.childOption(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
        bs2.group(group).childHandler(new ChannelInboundHandlerAdapter() {
            @Override
            public void channelActive(final ChannelHandlerContext ctx) throws Exception {
                ctx.channel().config().setAutoRead(false);
                Bootstrap bs = new Bootstrap();
                bs.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);

                bs.channel(EpollSocketChannel.class);
                bs.group(ctx.channel().eventLoop()).handler(new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelActive(ChannelHandlerContext context) throws Exception {
                        final EpollSocketChannel ch = (EpollSocketChannel) ctx.channel();
                        final EpollSocketChannel ch2 = (EpollSocketChannel) context.channel();
                        // We are splicing two channels together, at this point we have a tcp proxy which handles all
                        // the data transfer only in kernel space!

                        // Integer.MAX_VALUE will splice infinitely.
                        ch.spliceTo(ch2, Integer.MAX_VALUE).addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture future) throws Exception {
                                if (!future.isSuccess()) {
                                    future.channel().close();
                                }
                            }
                        });
                        // Trigger multiple splices to see if partial splicing works as well.
                        ch2.spliceTo(ch, SPLICE_LEN).addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture future) throws Exception {
                                if (!future.isSuccess()) {
                                    future.channel().close();
                                } else {
                                    ch2.spliceTo(ch, SPLICE_LEN).addListener(this);
                                }
                            }
                        });
                        ctx.channel().config().setAutoRead(true);
                    }

                    @Override
                    public void channelInactive(ChannelHandlerContext context) throws Exception {
                        context.close();
                    }
                });
                bs.connect(sc.localAddress()).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            ctx.close();
                        } else {
                            future.channel().closeFuture().addListener(new ChannelFutureListener() {
                                @Override
                                public void operationComplete(ChannelFuture future) throws Exception {
                                    ctx.close();
                                }
                            });
                        }