MessageForwarder1 Class — netty Architecture

Source Code

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java lines 416–469

    private static class MessageForwarder1 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 out = ctx.alloc().buffer(4);
            int m = ((Integer) msg).intValue();
            int expected = inCnt ++;
            assertEquals(expected, m);
            out.writeInt(m);

            ctx.fireChannelRead(out);
        }

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

            // Don't let the write request go to the server-side channel - just swallow.
            boolean swallow = this == ctx.pipeline().first();

            ByteBuf m = (ByteBuf) msg;
            int count = m.readableBytes() / 4;
            for (int j = 0; j < count; j ++) {
                int actual = m.readInt();
                int expected = outCnt ++;
                assertEquals(expected, actual);
                if (!swallow) {
                    ctx.write(actual);
                }
            }
            ctx.writeAndFlush(Unpooled.EMPTY_BUFFER, promise);
            m.release();
        }

        @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);
        }
    }