### Question:
ChannelSet implements ListenableAsyncCloseable { @Override public Completable closeAsyncGracefully() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { if (!stateUpdater.compareAndSet(ChannelSet.this, OPEN, GRACEFULLY_CLOSING)) { toSource(onClose).subscribe(subscriber); return; } if (channelMap.isEmpty()) { toSource(onClose).subscribe(subscriber); onCloseProcessor.onComplete(); return; } CompositeCloseable closeable = newCompositeCloseable().appendAll(() -> onClose); for (final Channel channel : channelMap.values()) { Attribute<AsyncCloseable> closeableAttribute = channel.attr(CHANNEL_CLOSEABLE_KEY); AsyncCloseable channelCloseable = closeableAttribute.getAndSet(null); if (channelCloseable != null) { closeable.merge(new AsyncCloseable() { @Override public Completable closeAsync() { return channelCloseable.closeAsync().publishOnOverride(immediate()); } @Override public Completable closeAsyncGracefully() { return channelCloseable.closeAsyncGracefully().publishOnOverride(immediate()); } }); } else { channel.close(); } } toSource(closeable.closeAsyncGracefully()).subscribe(subscriber); } }; } ChannelSet(Executor offloadingExecutor); boolean addIfAbsent(final Channel channel); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); @Override Completable onClose(); static final AttributeKey<AsyncCloseable> CHANNEL_CLOSEABLE_KEY; }### Answer:
@Test public void closeAsyncGracefullyWithNettyConnectionChannelHandler() throws Exception { Completable completable = closeAsyncGracefully(fixture, 100, SECONDS); verify(nettyConnection, never()).closeAsyncGracefully(); TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber(); toSource(completable).subscribe(subscriber); verify(nettyConnection).closeAsyncGracefully(); verify(channel, never()).close(); assertFalse(subscriber.pollTerminal(10, MILLISECONDS)); closeAsyncGracefullyCompletable.onComplete(); assertFalse(subscriber.pollTerminal(10, MILLISECONDS)); listener.operationComplete(channelCloseFuture); subscriber.awaitOnComplete(); }
@Test public void closeAsyncGracefullyWithoutNettyConnectionChannelHandler() throws InterruptedException { when(mockClosableAttribute.getAndSet(any())).thenReturn(null); Completable completable = closeAsyncGracefully(fixture, 100, SECONDS); verify(channel, never()).close(); TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber(); toSource(completable).subscribe(subscriber); verify(channel).close(); subscriber.awaitOnComplete(); }
@Test public void testCloseAsyncGracefullyTwiceTimesOut() throws Exception { Completable gracefulCompletable1 = closeAsyncGracefully(fixture, 100, MILLISECONDS); Completable gracefulCompletable2 = closeAsyncGracefully(fixture, 1000, MILLISECONDS); TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber(); toSource(gracefulCompletable1).subscribe(subscriber); verify(nettyConnection).closeAsyncGracefully(); TestCollectingCompletableSubscriber subscriber2 = new TestCollectingCompletableSubscriber(); toSource(gracefulCompletable2).subscribe(subscriber2); verify(nettyConnection, times(1)).closeAsyncGracefully(); gracefulCompletable1.toFuture().get(); subscriber.awaitOnComplete(); subscriber2.awaitOnComplete(); verify(channel).close(); }