class Connections {

    private final static InternalLogger LOG = InternalLoggerFactory.getInstance(Connections.class);

    private final Lock lock = new ReentrantLock();

    private final ClientResources clientResources;

    private final Map<RedisURI, StatefulRedisConnection<String, String>> connections;

    private volatile boolean closed = false;

    public Connections(ClientResources clientResources, Map<RedisURI, StatefulRedisConnection<String, String>> connections) {
        this.clientResources = clientResources;
        this.connections = connections;
    }

    /**
     * Add a connection for a {@link RedisURI}
     *
     * @param redisURI
     * @param connection
     */
    public void addConnection(RedisURI redisURI, StatefulRedisConnection<String, String> connection) {<FILL_FUNCTION_BODY>}

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

    /*
     * Initiate {@code CLUSTER NODES} on all connections and return the {@link Requests}.
     * @return the {@link Requests}.
     */
    public Requests requestTopology(long timeout, TimeUnit timeUnit) {

        return doRequest(() -> {

            CommandArgs<String, String> args = new CommandArgs<>(StringCodec.UTF8).add(CommandKeyword.NODES);
            Command<String, String, String> command = new Command<>(CommandType.CLUSTER, new StatusOutput<>(StringCodec.UTF8),
                    args);
            return new TimedAsyncCommand<>(command);
        }, timeout, timeUnit);
    }

    /*
     * Initiate {@code INFO} on all connections and return the {@link Requests}.
     * @return the {@link Requests}.
     */
    public Requests requestInfo(long timeout, TimeUnit timeUnit) {

        return doRequest(() -> {

            Command<String, String, String> command = new Command<>(CommandType.INFO, new StatusOutput<>(StringCodec.UTF8),
                    new CommandArgs<>(StringCodec.UTF8));
            return new TimedAsyncCommand<>(command);
        }, timeout, timeUnit);
    }

    /*
     * Initiate {@code CLUSTER NODES} on all connections and return the {@link Requests}.
     * @return the {@link Requests}.
     */
    private Requests doRequest(Supplier<TimedAsyncCommand<String, String, String>> commandFactory, long timeout,
            TimeUnit timeUnit) {

        Requests requests = new Requests();
        Duration timeoutDuration = Duration.ofNanos(timeUnit.toNanos(timeout));

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

                TimedAsyncCommand<String, String, String> timedCommand = commandFactory.get();

                clientResources.timer().newTimeout(it -> {
                    timedCommand.completeExceptionally(ExceptionFactory.createTimeoutException(timeoutDuration));
                }, timeout, timeUnit);

                entry.getValue().dispatch(timedCommand);
                requests.addRequest(entry.getKey(), timedCommand);
            }
        }
        finally {
            lock.unlock();
        }

        return requests;
    }

    public Connections retainAll(Set<RedisURI> connectionsToRetain) {

        Set<RedisURI> keys = new LinkedHashSet<>(connections.keySet());

        for (RedisURI key : keys) {
            if (!connectionsToRetain.contains(key)) {
                this.connections.remove(key);
            }
        }

        return this;
    }

}


        if (this.closed) { // fastpath
            connection.closeAsync();
            return;
        }

        try {
            lock.lock();
            if (this.closed) {
                connection.closeAsync();
                return;
            }

            this.connections.put(redisURI, connection);
        } finally {
            lock.unlock();
        }
    
<no_super_class>