class ReconnectionHandler {

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

    private static final Set<Class<?>> EXECUTION_EXCEPTION_TYPES = LettuceSets.unmodifiableSet(TimeoutException.class,
            CancellationException.class, RedisCommandTimeoutException.class, ConnectException.class);

    private final ClientOptions clientOptions;

    private final Bootstrap bootstrap;

    private final Mono<SocketAddress> socketAddressSupplier;

    private final ConnectionFacade connectionFacade;

    private volatile CompletableFuture<Channel> currentFuture;

    private volatile boolean reconnectSuspended;

    ReconnectionHandler(ClientOptions clientOptions, Bootstrap bootstrap, Mono<SocketAddress> socketAddressSupplier,
            Timer timer, ExecutorService reconnectWorkers, ConnectionFacade connectionFacade) {

        LettuceAssert.notNull(socketAddressSupplier, "SocketAddressSupplier must not be null");
        LettuceAssert.notNull(bootstrap, "Bootstrap must not be null");
        LettuceAssert.notNull(timer, "Timer must not be null");
        LettuceAssert.notNull(reconnectWorkers, "ExecutorService must not be null");
        LettuceAssert.notNull(connectionFacade, "ConnectionFacade must not be null");

        this.socketAddressSupplier = socketAddressSupplier;
        this.bootstrap = bootstrap;
        this.clientOptions = clientOptions;
        this.connectionFacade = connectionFacade;
    }

    /**
     * Initiate reconnect and return a {@link ChannelFuture} for synchronization. The resulting future either succeeds or fails.
     * It can be {@link ChannelFuture#cancel(boolean) canceled} to interrupt reconnection and channel initialization. A failed
     * {@link ChannelFuture} will close the channel.
     *
     * @return reconnect {@link ChannelFuture}.
     */
    protected Tuple2<CompletableFuture<Channel>, CompletableFuture<SocketAddress>> reconnect() {

        CompletableFuture<Channel> future = new CompletableFuture<>();
        CompletableFuture<SocketAddress> address = new CompletableFuture<>();

        socketAddressSupplier.subscribe(remoteAddress -> {

            address.complete(remoteAddress);

            if (future.isCancelled()) {
                return;
            }

            reconnect0(future, remoteAddress);

        }, ex -> {
            if (!address.isDone()) {
                address.completeExceptionally(ex);
            }
            future.completeExceptionally(ex);
        });

        this.currentFuture = future;
        return Tuples.of(future, address);
    }

    private void reconnect0(CompletableFuture<Channel> result, SocketAddress remoteAddress) {<FILL_FUNCTION_BODY>}

    boolean isReconnectSuspended() {
        return reconnectSuspended;
    }

    void setReconnectSuspended(boolean reconnectSuspended) {
        this.reconnectSuspended = reconnectSuspended;
    }

    void prepareClose() {

        CompletableFuture<?> currentFuture = this.currentFuture;
        if (currentFuture != null && !currentFuture.isDone()) {
            currentFuture.cancel(true);
        }
    }

    /**
     * @param throwable
     * @return {@code true} if {@code throwable} is an execution {@link Exception}.
     */
    public static boolean isExecutionException(Throwable throwable) {

        for (Class<?> type : EXECUTION_EXCEPTION_TYPES) {
            if (type.isAssignableFrom(throwable.getClass())) {
                return true;
            }
        }

        return false;
    }

    ClientOptions getClientOptions() {
        return clientOptions;
    }

}


        ChannelHandler handler = bootstrap.config().handler();

        // reinitialize SslChannelInitializer if Redis - SSL connection.
        if (SslConnectionBuilder.isSslChannelInitializer(handler)) {
            bootstrap.handler(SslConnectionBuilder.withSocketAddress(handler, remoteAddress));
        }

        ChannelFuture connectFuture = bootstrap.connect(remoteAddress);

        logger.debug("Reconnecting to Redis at {}", remoteAddress);

        result.whenComplete((c, t) -> {

            if (t instanceof CancellationException) {
                connectFuture.cancel(true);
            }
        });

        connectFuture.addListener(future -> {

            if (!future.isSuccess()) {
                result.completeExceptionally(future.cause());
                return;
            }

            RedisHandshakeHandler handshakeHandler = connectFuture.channel().pipeline().get(RedisHandshakeHandler.class);

            if (handshakeHandler == null) {
                result.completeExceptionally(new IllegalStateException("RedisHandshakeHandler not registered"));
                return;
            }

            handshakeHandler.channelInitialized().whenComplete((success, throwable) -> {

                if (throwable != null) {

                    if (isExecutionException(throwable)) {
                        result.completeExceptionally(throwable);
                        return;
                    }

                    if (clientOptions.isCancelCommandsOnReconnectFailure()) {
                        connectionFacade.reset();
                    }

                    if (clientOptions.isSuspendReconnectOnProtocolFailure()) {

                        logger.error("Disabling autoReconnect due to initialization failure", throwable);
                        setReconnectSuspended(true);
                    }

                    result.completeExceptionally(throwable);
                    return;
                }

                if (logger.isDebugEnabled()) {
                    logger.info("Reconnected to {}, Channel {}", remoteAddress,
                            ChannelLogDescriptor.logDescriptor(connectFuture.channel()));
                } else {
                    logger.info("Reconnected to {}", remoteAddress);
                }

                result.complete(connectFuture.channel());
            });

        });
    
<no_super_class>