### Question:
ChannelSet implements ListenableAsyncCloseable { @Override public Completable closeAsync() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { toSource(onClose).subscribe(subscriber); if (stateUpdater.getAndSet(ChannelSet.this, CLOSING) == CLOSING) { return; } if (channelMap.isEmpty()) { onCloseProcessor.onComplete(); return; } for (final Channel channel : channelMap.values()) { channel.close(); } } }; } 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 closeAsync() throws InterruptedException { Completable completable = fixture.closeAsync(); verify(channel, never()).close(); TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber(); toSource(completable).subscribe(subscriber); verify(channel).close(); subscriber.awaitOnComplete(); }