class SentinelTopologyProvider implements TopologyProvider {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(SentinelTopologyProvider.class);

    private final String masterId;

    private final RedisClient redisClient;

    private final RedisURI sentinelUri;

    private final Duration timeout;

    /**
     * Creates a new {@link SentinelTopologyProvider}.
     *
     * @param masterId must not be empty
     * @param redisClient must not be {@code null}.
     * @param sentinelUri must not be {@code null}.
     */
    public SentinelTopologyProvider(String masterId, RedisClient redisClient, RedisURI sentinelUri) {

        LettuceAssert.notEmpty(masterId, "MasterId must not be empty");
        LettuceAssert.notNull(redisClient, "RedisClient must not be null");
        LettuceAssert.notNull(sentinelUri, "Sentinel URI must not be null");

        this.masterId = masterId;
        this.redisClient = redisClient;
        this.sentinelUri = sentinelUri;
        this.timeout = sentinelUri.getTimeout();
    }

    @Override
    public List<RedisNodeDescription> getNodes() {

        logger.debug("lookup topology for masterId {}", masterId);

        try {
            return getNodesAsync().get(timeout.toMillis(), TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            throw Exceptions.bubble(e);
        }
    }

    @Override
    public CompletableFuture<List<RedisNodeDescription>> getNodesAsync() {<FILL_FUNCTION_BODY>}

    protected Mono<List<RedisNodeDescription>> getNodes(StatefulRedisSentinelConnection<String, String> connection) {

        RedisSentinelReactiveCommands<String, String> reactive = connection.reactive();

        Mono<Tuple2<Map<String, String>, List<Map<String, String>>>> masterAndReplicas = reactive.master(masterId)
                .zipWith(reactive.replicas(masterId).collectList()).timeout(this.timeout).flatMap(tuple -> {
                    return ResumeAfter.close(connection).thenEmit(tuple);
                }).doOnError(e -> connection.closeAsync());

        return masterAndReplicas.map(tuple -> {

            List<RedisNodeDescription> result = new ArrayList<>();

            result.add(toNode(tuple.getT1(), RedisInstance.Role.UPSTREAM));
            result.addAll(tuple.getT2().stream().filter(SentinelTopologyProvider::isAvailable)
                    .map(map -> toNode(map, RedisInstance.Role.REPLICA)).collect(Collectors.toList()));

            return result;
        });
    }

    private static boolean isAvailable(Map<String, String> map) {

        String flags = map.get("flags");
        if (flags != null) {
            if (flags.contains("s_down") || flags.contains("o_down") || flags.contains("disconnected")) {
                return false;
            }
        }
        return true;
    }

    private RedisNodeDescription toNode(Map<String, String> map, RedisInstance.Role role) {

        String ip = map.get("ip");
        String port = map.get("port");
        return new RedisMasterReplicaNode(ip, Integer.parseInt(port), sentinelUri, role);
    }

}


        logger.debug("lookup topology for masterId {}", masterId);

        Mono<StatefulRedisSentinelConnection<String, String>> connect = Mono
                .fromFuture(redisClient.connectSentinelAsync(StringCodec.UTF8, sentinelUri));

        return connect.flatMap(this::getNodes).toFuture();
    
<no_super_class>