SocketCancelWriteTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketCancelWriteTest.java lines 37–127

public class SocketCancelWriteTest extends AbstractSocketTest {

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

    public void testCancelWrite(ServerBootstrap sb, Bootstrap cb) throws Throwable {
        final TestHandler sh = new TestHandler();
        final TestHandler ch = new TestHandler();
        final ByteBuf a = Unpooled.buffer().writeByte('a');
        final ByteBuf b = Unpooled.buffer().writeByte('b');
        final ByteBuf c = Unpooled.buffer().writeByte('c');
        final ByteBuf d = Unpooled.buffer().writeByte('d');
        final ByteBuf e = Unpooled.buffer().writeByte('e');

        cb.handler(ch);
        sb.childHandler(sh);

        Channel sc = sb.bind().sync().channel();
        Channel cc = cb.connect(sc.localAddress()).sync().channel();

        ChannelFuture f = cc.write(a);
        assertTrue(f.cancel(false));
        cc.writeAndFlush(b);
        cc.write(c);
        ChannelFuture f2 = cc.write(d);
        assertTrue(f2.cancel(false));
        cc.writeAndFlush(e);

        while (sh.counter < 3) {
            if (sh.exception.get() != null) {
                break;
            }
            if (ch.exception.get() != null) {
                break;
            }
            Thread.sleep(50);
        }
        sh.channel.close().sync();
        ch.channel.close().sync();
        sc.close().sync();

        if (sh.exception.get() != null && !(sh.exception.get() instanceof IOException)) {
            throw sh.exception.get();
        }
        if (sh.exception.get() != null) {
            throw sh.exception.get();
        }
        if (ch.exception.get() != null && !(ch.exception.get() instanceof IOException)) {
            throw ch.exception.get();
        }
        if (ch.exception.get() != null) {
            throw ch.exception.get();
        }
        assertEquals(0, ch.counter);
        assertEquals(Unpooled.wrappedBuffer(new byte[]{'b', 'c', 'e'}), sh.received);
    }

    private static class TestHandler extends SimpleChannelInboundHandler<ByteBuf> {
        volatile Channel channel;
        final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
        volatile int counter;
        final ByteBuf received = Unpooled.buffer();
        @Override
        public void channelActive(ChannelHandlerContext ctx)
                throws Exception {
            channel = ctx.channel();
        }

        @Override
        public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
            counter += in.readableBytes();
            received.writeBytes(in);
        }