class 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) {
                        // Upon shutdown of the set, we will close all live channels. If close of individual channels
                        // are offloaded, then this would trigger a surge in threads required to offload these closures.
                        // Here we assume that if there is any offloading required, it is done by offloading the
                        // Completable returned by closeAsyncGracefully() hence offloading each channel is not required.
                        // Hence, we override the offloading on each channel for this particular subscribe to use
                        // immediate and effectively disable offloading on each channel.
                        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);
            }
        };
    }

    public  ChannelSet(Executor offloadingExecutor);

}