class 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()) {
                    // We don't try to catch exceptions here because we're only invoking Netty or ServiceTalk code, no
                    // user-provided code.
                    channel.close();
                }
            }
        };
    }

    public  ChannelSet(Executor offloadingExecutor);

}