class RedisHandshakeHandler extends ChannelInboundHandlerAdapter {

    private final ConnectionInitializer connectionInitializer;

    private final ClientResources clientResources;

    private final Duration initializeTimeout;

    private final CompletableFuture<Void> handshakeFuture = new CompletableFuture<>();

    private volatile boolean timedOut = false;

    public RedisHandshakeHandler(ConnectionInitializer connectionInitializer, ClientResources clientResources,
            Duration initializeTimeout) {
        this.connectionInitializer = connectionInitializer;
        this.clientResources = clientResources;
        this.initializeTimeout = initializeTimeout;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {

        Runnable timeoutGuard = () -> {

            timedOut = true;
            if (handshakeFuture.isDone()) {
                return;
            }

            fail(ctx, new RedisCommandTimeoutException(
                    "Connection initialization timed out after " + ExceptionFactory.formatTimeout(initializeTimeout)));
        };

        Timeout timeoutHandle = clientResources.timer().newTimeout(t -> {

            if (clientResources.eventExecutorGroup().isShuttingDown()) {
                timeoutGuard.run();
                return;
            }

            clientResources.eventExecutorGroup().submit(timeoutGuard);
        }, initializeTimeout.toNanos(), TimeUnit.NANOSECONDS);

        handshakeFuture.thenAccept(ignore -> {
            timeoutHandle.cancel();
        });

        super.channelRegistered(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        if (shouldFail()) {
            fail(ctx, new RedisConnectionException("Connection closed prematurely"));
        }

        super.channelInactive(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {<FILL_FUNCTION_BODY>}

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        if (shouldFail()) {
            fail(ctx, cause);
        }

        super.exceptionCaught(ctx, cause);
    }

    /**
     * Complete the handshake future successfully.
     */
    protected void succeed() {
        handshakeFuture.complete(null);
    }

    /**
     * Complete the handshake future with an error and close the channel..
     */
    protected void fail(ChannelHandlerContext ctx, Throwable cause) {

        ctx.close().addListener(closeFuture -> {
            handshakeFuture.completeExceptionally(cause);
        });
    }

    /**
     * @return future to synchronize channel initialization. Returns a new future for every reconnect.
     */
    public CompletionStage<Void> channelInitialized() {
        return handshakeFuture;
    }

    private boolean shouldFail() {
        return !handshakeFuture.isDone() && !timedOut;
    }

}


        CompletionStage<Void> future = connectionInitializer.initialize(ctx.channel());

        future.whenComplete((ignore, throwable) -> {

            if (throwable != null) {

                if (shouldFail()) {
                    fail(ctx, throwable);
                }
            } else {
                ctx.fireChannelActive();
                succeed();
            }
        });
    
<no_super_class>