@Test public void testPeerWriteInWritePromiseCompleteSameEventLoopPreservesOrder() throws InterruptedException { Bootstrap cb = new Bootstrap(); ServerBootstrap sb = new ServerBootstrap(); final CountDownLatch messageLatch = new CountDownLatch(2); final ByteBuf data = Unpooled.wrappedBuffer(new byte[1024]); final ByteBuf data2 = Unpooled.wrappedBuffer(new byte[512]); final CountDownLatch serverChannelLatch = new CountDownLatch(1); final AtomicReference<Channel> serverChannelRef = new AtomicReference<Channel>(); try { cb.group(sharedGroup) .channel(LocalChannel.class) .handler(new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (data2.equals(msg) && messageLatch.getCount() == 1) { ReferenceCountUtil.safeRelease(msg); messageLatch.countDown(); } else { super.channelRead(ctx, msg); } } }); sb.group(sharedGroup) .channel(LocalServerChannel.class) .childHandler(new ChannelInitializer<LocalChannel>() { @Override public void initChannel(LocalChannel ch) throws Exception { ch.pipeline().addLast(new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (data.equals(msg) && messageLatch.getCount() == 2) { ReferenceCountUtil.safeRelease(msg); messageLatch.countDown(); } else { super.channelRead(ctx, msg); } } }); serverChannelRef.set(ch); serverChannelLatch.countDown(); } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).syncUninterruptibly().channel(); cc = cb.connect(sc.localAddress()).syncUninterruptibly().channel(); assertTrue(serverChannelLatch.await(5, SECONDS)); final Channel ccCpy = cc; cc.pipeline().lastContext().executor().execute(new Runnable() { @Override public void run() { ChannelPromise promise = ccCpy.newPromise(); promise.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { Channel serverChannelCpy = serverChannelRef.get(); serverChannelCpy.writeAndFlush( data2.retainedDuplicate(), serverChannelCpy.newPromise()); } }); ccCpy.writeAndFlush(data.retainedDuplicate(), promise); } }); assertTrue(messageLatch.await(5, SECONDS)); } finally { closeChannel(cc); closeChannel(sc); } } finally { data.release(); data2.release(); } }
@Override public LocalAddress localAddress() { return (LocalAddress) super.localAddress(); }
LocalChannel extends AbstractChannel { @Override public LocalAddress localAddress() { return (LocalAddress) super.localAddress(); } }
LocalChannel extends AbstractChannel { @Override public LocalAddress localAddress() { return (LocalAddress) super.localAddress(); } LocalChannel(); protected  LocalChannel(LocalServerChannel parent, LocalChannel peer); }
LocalChannel extends AbstractChannel { @Override public LocalAddress localAddress() { return (LocalAddress) super.localAddress(); } LocalChannel(); protected  LocalChannel(LocalServerChannel parent, LocalChannel peer); @Override ChannelMetadata metadata(); @Override ChannelConfig config(); @Override LocalServerChannel parent(); @Override LocalAddress localAddress(); @Override LocalAddress remoteAddress(); @Override boolean isOpen(); @Override boolean isActive(); }
LocalChannel extends AbstractChannel { @Override public LocalAddress localAddress() { return (LocalAddress) super.localAddress(); } LocalChannel(); protected  LocalChannel(LocalServerChannel parent, LocalChannel peer); @Override ChannelMetadata metadata(); @Override ChannelConfig config(); @Override LocalServerChannel parent(); @Override LocalAddress localAddress(); @Override LocalAddress remoteAddress(); @Override boolean isOpen(); @Override boolean isActive();  }