class Connections extends CompletableEventLatchSupport<Tuple2<RedisURI, StatefulRedisConnection<String, String>>, Connections>
        implements AsyncCloseable {

    private final Lock lock = new ReentrantLock();
    private final Map<RedisURI, StatefulRedisConnection<String, String>> connections = new TreeMap<>(
            ReplicaUtils.RedisURIComparator.INSTANCE);

    private final List<Throwable> exceptions = new CopyOnWriteArrayList<>();

    private final List<RedisNodeDescription> nodes;

    private volatile boolean closed = false;

    public Connections(int expectedConnectionCount, List<RedisNodeDescription> nodes) {
        super(expectedConnectionCount);
        this.nodes = nodes;
    }

    @Override
    protected void onAccept(Tuple2<RedisURI, StatefulRedisConnection<String, String>> value) {

        if (this.closed) {
            value.getT2().closeAsync();
            return;
        }

        try {
            lock.lock();
            this.connections.put(value.getT1(), value.getT2());
        } finally {
            lock.unlock();
        }
    }

    @Override
    protected void onError(Throwable value) {
        this.exceptions.add(value);
    }

    @Override
    protected void onDrop(Tuple2<RedisURI, StatefulRedisConnection<String, String>> value) {
        value.getT2().closeAsync();
    }

    @Override
    protected void onDrop(Throwable value) {

    }

    @Override
    protected void onEmit(Emission<Connections> emission) {

        if (getExpectedCount() != 0 && this.connections.isEmpty() && !this.exceptions.isEmpty()) {

            RedisConnectionException collector = new RedisConnectionException(
                    "Unable to establish a connection to Redis Master/Replica");
            this.exceptions.forEach(collector::addSuppressed);

            emission.error(collector);
        } else {
            emission.success(this);
        }
    }

    /**
     * @return {@code true} if no connections present.
     */
    public boolean isEmpty() {
        try {
            lock.lock();
            return this.connections.isEmpty();
        }
        finally {
            lock.unlock();
        }
    }

    /*
     * Initiate {@code PING} on all connections and return the {@link Requests}.
     * @return the {@link Requests}.
     */
    public Requests requestPing() {

        Set<Map.Entry<RedisURI, StatefulRedisConnection<String, String>>> entries = new LinkedHashSet<>(
                this.connections.entrySet());
        Requests requests = new Requests(entries.size(), this.nodes);

        for (Map.Entry<RedisURI, StatefulRedisConnection<String, String>> entry : entries) {

            CommandArgs<String, String> args = new CommandArgs<>(StringCodec.ASCII).add(CommandKeyword.NODES);
            Command<String, String, String> command = new Command<>(CommandType.PING, new StatusOutput<>(StringCodec.ASCII),
                    args);
            TimedAsyncCommand<String, String, String> timedCommand = new TimedAsyncCommand<>(command);

            entry.getValue().dispatch(timedCommand);
            requests.addRequest(entry.getKey(), timedCommand);
        }

        return requests;
    }

    /**
     * Close all connections.
     */
    public CompletableFuture<Void> closeAsync() {<FILL_FUNCTION_BODY>}

}


        List<CompletableFuture<?>> close = new ArrayList<>(this.connections.size());
        List<RedisURI> toRemove = new ArrayList<>(this.connections.size());

        this.closed = true;

        for (Map.Entry<RedisURI, StatefulRedisConnection<String, String>> entry : this.connections.entrySet()) {

            toRemove.add(entry.getKey());
            close.add(entry.getValue().closeAsync());
        }

        for (RedisURI redisURI : toRemove) {
            this.connections.remove(redisURI);
        }

        return Futures.allOf(close);
    
<methods>public void <init>(int) ,public final void accept(Tuple2<io.lettuce.core.RedisURI,StatefulRedisConnection<java.lang.String,java.lang.String>>) ,public final void accept(java.lang.Throwable) ,public final int getExpectedCount() ,public final CompletionStage<io.lettuce.core.masterreplica.Connections> getOrTimeout(java.time.Duration, java.util.concurrent.ScheduledExecutorService) <variables>private static final int GATE_CLOSED,private static final int GATE_OPEN,private static final AtomicIntegerFieldUpdater<CompletableEventLatchSupport#RAW> GATE_UPDATER,private static final AtomicIntegerFieldUpdater<CompletableEventLatchSupport#RAW> NOTIFICATIONS_UPDATER,private final non-sealed int expectedCount,private volatile int gate,private volatile int notifications,private final CompletableFuture<io.lettuce.core.masterreplica.Connections> selfFuture,private volatile ScheduledFuture<?> timeoutScheduleFuture