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();  }
@Test public void testLocalAddressReuse() throws Exception { for (int i = 0; i < 2; i ++) { Bootstrap cb = new Bootstrap(); ServerBootstrap sb = new ServerBootstrap(); cb.group(group1) .channel(LocalChannel.class) .handler(new TestHandler()); sb.group(group2) .channel(LocalServerChannel.class) .childHandler(new ChannelInitializer<LocalChannel>() { @Override public void initChannel(LocalChannel ch) throws Exception { ch.pipeline().addLast(new TestHandler()); } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).sync().channel(); final CountDownLatch latch = new CountDownLatch(1); cc = cb.connect(sc.localAddress()).sync().channel(); final Channel ccCpy = cc; cc.eventLoop().execute(new Runnable() { @Override public void run() { ccCpy.pipeline().fireChannelRead("Hello, World"); latch.countDown(); } }); assertTrue(latch.await(5, SECONDS)); closeChannel(cc); closeChannel(sc); sc.closeFuture().sync(); assertNull(String.format( "Expected null, got channel '%s' for local address '%s'", LocalChannelRegistry.get(TEST_ADDRESS), TEST_ADDRESS), LocalChannelRegistry.get(TEST_ADDRESS)); } finally { closeChannel(cc); closeChannel(sc); } } }
@Test public void testWriteFailsFastOnClosedChannel() throws Exception { Bootstrap cb = new Bootstrap(); ServerBootstrap sb = new ServerBootstrap(); cb.group(group1) .channel(LocalChannel.class) .handler(new TestHandler()); sb.group(group2) .channel(LocalServerChannel.class) .childHandler(new ChannelInitializer<LocalChannel>() { @Override public void initChannel(LocalChannel ch) throws Exception { ch.pipeline().addLast(new TestHandler()); } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).sync().channel(); cc = cb.connect(sc.localAddress()).sync().channel(); cc.close().sync(); try { cc.writeAndFlush(new Object()).sync(); fail("must raise a ClosedChannelException"); } catch (Exception e) { assertThat(e, is(instanceOf(ClosedChannelException.class))); if (e.getStackTrace().length > 0) { assertThat( e.getStackTrace()[0].getClassName(), is(AbstractChannel.class.getName() + "$AbstractUnsafe")); e.printStackTrace(); } } } finally { closeChannel(cc); closeChannel(sc); } }
@Test public void testServerCloseChannelSameEventLoop() throws Exception { final CountDownLatch latch = new CountDownLatch(1); ServerBootstrap sb = new ServerBootstrap() .group(group2) .channel(LocalServerChannel.class) .childHandler(new SimpleChannelInboundHandler<Object>() { @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { ctx.close(); latch.countDown(); } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).sync().channel(); Bootstrap b = new Bootstrap() .group(group2) .channel(LocalChannel.class) .handler(new SimpleChannelInboundHandler<Object>() { @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { } }); cc = b.connect(sc.localAddress()).sync().channel(); cc.writeAndFlush(new Object()); assertTrue(latch.await(5, SECONDS)); } finally { closeChannel(cc); closeChannel(sc); } }
@Test public void localChannelRaceCondition() throws Exception { final CountDownLatch closeLatch = new CountDownLatch(1); final EventLoopGroup clientGroup = new DefaultEventLoopGroup(1) { @Override protected EventLoop newChild(Executor threadFactory, Object... args) throws Exception { return new SingleThreadEventLoop(this, threadFactory, true) { @Override protected void run() { for (;;) { Runnable task = takeTask(); if (task != null) { if (task.getClass().getEnclosingClass() == LocalChannel.class) { try { closeLatch.await(); } catch (InterruptedException e) { throw new Error(e); } } task.run(); updateLastExecutionTime(); } if (confirmShutdown()) { break; } } } }; } }; Channel sc = null; Channel cc = null; try { ServerBootstrap sb = new ServerBootstrap(); sc = sb.group(group2). channel(LocalServerChannel.class). childHandler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) throws Exception { ch.close(); closeLatch.countDown(); } }). bind(TEST_ADDRESS). sync().channel(); Bootstrap bootstrap = new Bootstrap(); bootstrap.group(clientGroup). channel(LocalChannel.class). handler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) throws Exception { } }); ChannelFuture future = bootstrap.connect(sc.localAddress()); assertTrue("Connection should finish, not time out", future.await(200)); cc = future.channel(); } finally { closeChannel(cc); closeChannel(sc); clientGroup.shutdownGracefully(0, 0, SECONDS).await(); } }
@Test public void testReRegister() { Bootstrap cb = new Bootstrap(); ServerBootstrap sb = new ServerBootstrap(); cb.group(group1) .channel(LocalChannel.class) .handler(new TestHandler()); sb.group(group2) .channel(LocalServerChannel.class) .childHandler(new ChannelInitializer<LocalChannel>() { @Override public void initChannel(LocalChannel ch) throws Exception { ch.pipeline().addLast(new TestHandler()); } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).syncUninterruptibly().channel(); cc = cb.connect(sc.localAddress()).syncUninterruptibly().channel(); cc.deregister().syncUninterruptibly(); } finally { closeChannel(cc); closeChannel(sc); } }
@Test public void testWriteInWritePromiseCompletePreservesOrder() 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]); try { cb.group(group1) .channel(LocalChannel.class) .handler(new TestHandler()); sb.group(group2) .channel(LocalServerChannel.class) .childHandler(new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { final long count = messageLatch.getCount(); if ((data.equals(msg) && count == 2) || (data2.equals(msg) && count == 1)) { ReferenceCountUtil.safeRelease(msg); messageLatch.countDown(); } else { super.channelRead(ctx, msg); } } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).syncUninterruptibly().channel(); cc = cb.connect(sc.localAddress()).syncUninterruptibly().channel(); 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 { ccCpy.writeAndFlush(data2.retainedDuplicate(), ccCpy.newPromise()); } }); ccCpy.writeAndFlush(data.retainedDuplicate(), promise); } }); assertTrue(messageLatch.await(5, SECONDS)); } finally { closeChannel(cc); closeChannel(sc); } } finally { data.release(); data2.release(); } }
@Test public void testPeerWriteInWritePromiseCompleteDifferentEventLoopPreservesOrder() 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>(); cb.group(group1) .channel(LocalChannel.class) .handler(new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (data2.equals(msg)) { ReferenceCountUtil.safeRelease(msg); messageLatch.countDown(); } else { super.channelRead(ctx, msg); } } }); sb.group(group2) .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)) { 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); data.release(); data2.release(); } }
@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(); } }
@Test(timeout = 3000) public void testConnectFutureBeforeChannelActive() throws Exception { Bootstrap cb = new Bootstrap(); ServerBootstrap sb = new ServerBootstrap(); cb.group(group1) .channel(LocalChannel.class) .handler(new ChannelInboundHandlerAdapter()); sb.group(group2) .channel(LocalServerChannel.class) .childHandler(new ChannelInitializer<LocalChannel>() { @Override public void initChannel(LocalChannel ch) throws Exception { ch.pipeline().addLast(new TestHandler()); } }); Channel sc = null; Channel cc = null; try { sc = sb.bind(TEST_ADDRESS).sync().channel(); cc = cb.register().sync().channel(); final ChannelPromise promise = cc.newPromise(); final Promise<Void> assertPromise = cc.eventLoop().newPromise(); cc.pipeline().addLast(new TestHandler() { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { if (promise.isDone()) { assertPromise.setSuccess(null); } else { assertPromise.setFailure(new AssertionError("connect promise should be done")); } } }); cc.connect(sc.localAddress(), promise).sync(); assertPromise.syncUninterruptibly(); assertTrue(promise.isSuccess()); } finally { closeChannel(cc); closeChannel(sc); } }
@Test public void testNotLeakBuffersWhenCloseByRemotePeer() throws Exception { Bootstrap cb = new Bootstrap(); ServerBootstrap sb = new ServerBootstrap(); cb.group(sharedGroup) .channel(LocalChannel.class) .handler(new SimpleChannelInboundHandler<ByteBuf>() { @Override public void channelActive(final ChannelHandlerContext ctx) throws Exception { ctx.writeAndFlush(ctx.alloc().buffer().writeZero(100)); } @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception { } }); sb.group(sharedGroup) .channel(LocalServerChannel.class) .childHandler(new ChannelInitializer<LocalChannel>() { @Override public void initChannel(LocalChannel ch) throws Exception { ch.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() { @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception { while (buffer.isReadable()) { ctx.write(buffer.readRetainedSlice(1)); } ctx.flush(); ctx.close(); } }); } }); Channel sc = null; LocalChannel cc = null; try { sc = sb.bind(TEST_ADDRESS).sync().channel(); cc = (LocalChannel) cb.connect(sc.localAddress()).sync().channel(); closeChannel(cc); assertTrue(cc.inboundBuffer.isEmpty()); closeChannel(sc); } finally { closeChannel(cc); closeChannel(sc); } }