docs: public CompletableFuture<StatefulRedisConnection<K, V>> getConnectionAsync(Intent intent) {
  
          if (debugEnabled) {
              logger.debug("getConnectionAsync(" + intent + ")");
          }
  
          if (readFrom != null && intent == Intent.READ) {
              List<RedisNodeDescription> selection = readFrom.select(new ReadFrom.Nodes() {
                  @Override
                  public List<RedisNodeDescription> getNodes() {
                      return knownNodes;
                  }
  
                  @Override
                  public Iterator<RedisNodeDescription> iterator() {
                      return knownNodes.iterator();
                  }
              });
  
              if (selection.isEmpty()) {
                  throw new RedisException(String.format("Cannot determine a node to read (Known nodes: %s) with setting %s",
                          knownNodes, readFrom));
              }
  
              try {
  
                  Flux<StatefulRedisConnection<K, V>> connections = Flux.empty();
  
                  for (RedisNodeDescription node : selection) {
                      connections = connections.concatWith(Mono.fromFuture(getConnection(node)));
                  }
  
                  if (OrderingReadFromAccessor.isOrderSensitive(readFrom) || selection.size() == 1) {
                      return connections.filter(StatefulConnection::isOpen).next().switchIfEmpty(connections.next()).toFuture();
                  }
  
                  return connections.filter(StatefulConnection::isOpen).collectList().map(it -> {
                      int index = ThreadLocalRandom.current().nextInt(it.size());
                      return it.get(index);
                  }).switchIfEmpty(connections.next()).toFuture();
              } catch (RuntimeException e) {
                  throw Exceptions.bubble(e);
              }
          }
  
          return getConnection(getMaster());
      }