private <K, V, T extends RedisChannelHandler<K, V>, S> ConnectionFuture<S> connectStatefulAsync(T connection,
            RedisCodec<K, V> codec,
            DefaultEndpoint endpoint, RedisURI connectionSettings, Mono<SocketAddress> socketAddressSupplier,
            Supplier<CommandHandler> commandHandlerSupplier) {

        ConnectionBuilder connectionBuilder = createConnectionBuilder(connection, endpoint, connectionSettings,
                socketAddressSupplier, commandHandlerSupplier);

        if (clientOptions.isPingBeforeActivateConnection()) {
            if (hasPassword(connectionSettings)) {
                connectionBuilder.enableAuthPingBeforeConnect();
            } else {
                connectionBuilder.enablePingBeforeConnect();
            }
        }

        ConnectionFuture<RedisChannelHandler<K, V>> future = initializeChannelAsync(connectionBuilder);
        ConnectionFuture<?> sync = future;

        if (!clientOptions.isPingBeforeActivateConnection() && hasPassword(connectionSettings)) {

            sync = sync.thenCompose(channelHandler -> {

                CommandArgs<K, V> args = new CommandArgs<>(codec).add(connectionSettings.getPassword());
                AsyncCommand<K, V, String> command = new AsyncCommand<>(new Command<>(CommandType.AUTH, new StatusOutput<>(
                        codec), args));

                if (connection instanceof StatefulRedisClusterConnectionImpl) {
                    ((StatefulRedisClusterConnectionImpl) connection).dispatch(command);
                }

                if (connection instanceof StatefulRedisConnectionImpl) {
                    ((StatefulRedisConnectionImpl) connection).dispatch(command);
                }

                return command;
            });

        }

        if (LettuceStrings.isNotEmpty(connectionSettings.getClientName())) {
            sync = sync.thenApply(channelHandler -> {

                if (connection instanceof StatefulRedisClusterConnectionImpl) {
                    ((StatefulRedisClusterConnectionImpl) connection).setClientName(connectionSettings.getClientName());
                }

                if (connection instanceof StatefulRedisConnectionImpl) {
                    ((StatefulRedisConnectionImpl) connection).setClientName(connectionSettings.getClientName());
                }
                return channelHandler;
            });
        }

        return sync.thenApply(channelHandler -> (S) connection);
    }
private < K , V , T extends RedisChannelHandler < K , V > , S > ConnectionFuture < S > connectStatefulAsync ( T connection , RedisCodec < K , V > codec , DefaultEndpoint endpoint , RedisURI connectionSettings , Mono < SocketAddress > socketAddressSupplier , Supplier < CommandHandler > commandHandlerSupplier ) { ConnectionBuilder connectionBuilder = createConnectionBuilder ( connection , endpoint , connectionSettings , socketAddressSupplier , commandHandlerSupplier ) ; if ( clientOptions . isPingBeforeActivateConnection ( ) ) { if ( hasPassword ( connectionSettings ) ) { connectionBuilder . enableAuthPingBeforeConnect ( ) ; } else { connectionBuilder . enablePingBeforeConnect ( ) ; } } ConnectionFuture < RedisChannelHandler < K , V > > future = initializeChannelAsync ( connectionBuilder ) ; ConnectionFuture < ? > sync = future ; if ( ! clientOptions . isPingBeforeActivateConnection ( ) && hasPassword ( connectionSettings ) ) { sync = sync . thenCompose ( channelHandler -> { CommandArgs < K , V > args = new CommandArgs < > ( codec ) . add ( connectionSettings . getPassword ( ) ) ; AsyncCommand < K , V , String > command = new AsyncCommand < > ( new Command < > ( CommandType . AUTH , new StatusOutput < > ( codec ) , args ) ) ; if ( connection instanceof StatefulRedisClusterConnectionImpl ) { ( ( StatefulRedisClusterConnectionImpl ) connection ) . dispatch ( command ) ; } if ( connection instanceof StatefulRedisConnectionImpl ) { ( ( StatefulRedisConnectionImpl ) connection ) . dispatch ( command ) ; } return command ; } ) ; } if ( LettuceStrings . isNotEmpty ( connectionSettings . getClientName ( ) ) ) { sync = sync . thenApply ( channelHandler -> { if ( connection instanceof StatefulRedisClusterConnectionImpl ) { ( ( StatefulRedisClusterConnectionImpl ) connection ) . setClientName ( connectionSettings . getClientName ( ) ) ; } if ( connection instanceof StatefulRedisConnectionImpl ) { ( ( StatefulRedisConnectionImpl ) connection ) . setClientName ( connectionSettings . getClientName ( ) ) ; } return channelHandler ; } ) ; } return sync . thenApply ( channelHandler -> ( S ) connection ) ; }
Initiates a channel connection considering {@link ClientOptions} initialization options, authentication and client name
options.