@Test public void shouldNotPropagateDownstreamCloseWhileBarrierFutureNotDone() throws Exception { context.checking(new Expectations() { { oneOf(upstream).handleUpstream(with(any(ChannelHandlerContext.class)), with(any(PreparationEvent.class))); oneOf(upstream).handleUpstream(with(any(ChannelHandlerContext.class)), with(channelState(OPEN, TRUE))); } }); Channel channel = channelFactory.newChannel(pipeline); close(channel); ChannelFuture handlerFuture = handler.getHandlerFuture(); assertFalse(handlerFuture.isDone()); assertTrue(handler.hasQueuedChannelEvents()); context.assertIsSatisfied(); }
boolean hasQueuedChannelEvents() { return queue != null && !queue.isEmpty(); }
AwaitBarrierDownstreamHandler extends AbstractBarrierHandler implements ChannelDownstreamHandler { boolean hasQueuedChannelEvents() { return queue != null && !queue.isEmpty(); } }
AwaitBarrierDownstreamHandler extends AbstractBarrierHandler implements ChannelDownstreamHandler { boolean hasQueuedChannelEvents() { return queue != null && !queue.isEmpty(); } AwaitBarrierDownstreamHandler(Barrier barrier); }
AwaitBarrierDownstreamHandler extends AbstractBarrierHandler implements ChannelDownstreamHandler { boolean hasQueuedChannelEvents() { return queue != null && !queue.isEmpty(); } AwaitBarrierDownstreamHandler(Barrier barrier); @Override void prepareRequested(final ChannelHandlerContext ctx, PreparationEvent evt); @Override void handleDownstream(ChannelHandlerContext ctx, ChannelEvent evt); }
AwaitBarrierDownstreamHandler extends AbstractBarrierHandler implements ChannelDownstreamHandler { boolean hasQueuedChannelEvents() { return queue != null && !queue.isEmpty(); } AwaitBarrierDownstreamHandler(Barrier barrier); @Override void prepareRequested(final ChannelHandlerContext ctx, PreparationEvent evt); @Override void handleDownstream(ChannelHandlerContext ctx, ChannelEvent evt);  }