private < K , V > CompletableFuture < StatefulRedisClusterPubSubConnection < K , V > > connectClusterPubSubAsync ( RedisCodec < K , V > codec ) { if ( partitions == null ) { return Futures . failed ( new IllegalStateException ( "Partitions not initialized. Initialize via RedisClusterClient.getPartitions()." ) ) ; } activateTopologyRefreshIfNeeded ( ) ; logger . debug ( "connectClusterPubSub(" + initialUris + ")" ) ; Mono < SocketAddress > socketAddressSupplier = getSocketAddressSupplier ( TopologyComparators :: sortByClientCount ) ; PubSubClusterEndpoint < K , V > endpoint = new PubSubClusterEndpoint < K , V > ( clientOptions , clientResources ) ; RedisChannelWriter writer = endpoint ; if ( CommandExpiryWriter . isSupported ( clientOptions ) ) { writer = new CommandExpiryWriter ( writer , clientOptions , clientResources ) ; } ClusterDistributionChannelWriter clusterWriter = new ClusterDistributionChannelWriter ( clientOptions , writer , clusterTopologyRefreshScheduler ) ; StatefulRedisClusterPubSubConnectionImpl < K , V > connection = new StatefulRedisClusterPubSubConnectionImpl <> ( endpoint , clusterWriter , codec , timeout ) ; ClusterPubSubConnectionProvider < K , V > pooledClusterConnectionProvider = new ClusterPubSubConnectionProvider <> ( this , clusterWriter , codec , connection . getUpstreamListener ( ) , clusterTopologyRefreshScheduler ) ; clusterWriter . setClusterConnectionProvider ( pooledClusterConnectionProvider ) ; connection . setPartitions ( partitions ) ; Supplier < CommandHandler > commandHandlerSupplier = ( ) -> new PubSubCommandHandler <> ( clientOptions , clientResources , codec , endpoint ) ; Mono < StatefulRedisClusterPubSubConnectionImpl < K , V > > connectionMono = Mono . defer ( ( ) -> connect ( socketAddressSupplier , codec , endpoint , connection , commandHandlerSupplier ) ) ; for ( int i = 1 ; i < getConnectionAttempts ( ) ; i ++ ) { connectionMono = connectionMono . onErrorResume ( t -> connect ( socketAddressSupplier , codec , endpoint , connection , commandHandlerSupplier ) ) ; } return connectionMono . flatMap ( c -> c . reactive ( ) . command ( ) . collectList ( ) // . map ( CommandDetailParser :: parse ) // . doOnNext ( detail -> c . setState ( new RedisState ( detail ) ) ) . doOnError ( e -> c . setState ( new RedisState ( Collections . emptyList ( ) ) ) ) . then ( Mono . just ( c ) ) . onErrorResume ( RedisCommandExecutionException . class , e -> Mono . just ( c ) ) ) . doOnNext ( c -> connection . registerCloseables ( closeableResources , clusterWriter , pooledClusterConnectionProvider ) ) . map ( it -> ( StatefulRedisClusterPubSubConnection < K , V > ) it ) . toFuture ( ) ; }
Create a clustered connection with command distributor .