MessageForwarder2 Class — netty Architecture

Source Code

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java lines 474–521

    private static class MessageForwarder2 extends ChannelDuplexHandler {

        private final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
        private volatile int inCnt;
        private volatile int outCnt;
        private volatile Thread t;

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            Thread t = this.t;
            if (t == null) {
                this.t = Thread.currentThread();
            } else {
                assertSame(t, Thread.currentThread());
            }

            ByteBuf m = (ByteBuf) msg;
            int count = m.readableBytes() / 4;
            for (int j = 0; j < count; j ++) {
                int actual = m.readInt();
                int expected = inCnt ++;
                assertEquals(expected, actual);
                ctx.fireChannelRead(actual);
            }
            m.release();
        }

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            assertSame(t, Thread.currentThread());

            ByteBuf out = ctx.alloc().buffer(4);
            int m = (Integer) msg;
            int expected = outCnt ++;
            assertEquals(expected, m);
            out.writeInt(m);

            ctx.write(out, promise);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            exception.compareAndSet(null, cause);
            //System.err.print("[" + Thread.currentThread().getName() + "] ");
            //cause.printStackTrace();
            super.exceptionCaught(ctx, cause);
        }
    }