CompletableFuture < Channel > getChannel ( Address address , String messageType ) { List < CompletableFuture < Channel >> channelPool = getChannelPool ( address ) ; int offset = getChannelOffset ( messageType ) ; CompletableFuture < Channel > channelFuture = channelPool . get ( offset ) ; if ( channelFuture == null || channelFuture . isCompletedExceptionally ( ) ) { synchronized ( channelPool ) { channelFuture = channelPool . get ( offset ) ; if ( channelFuture == null || channelFuture . isCompletedExceptionally ( ) ) { LOGGER . debug ( "Connecting to {}" , address ) ; channelFuture = factory . apply ( address ) ; channelFuture . whenComplete ( ( channel , error ) -> { if ( error == null ) { LOGGER . debug ( "Connected to {}" , channel . remoteAddress ( ) ) ; } else { LOGGER . debug ( "Failed to connect to {}" , address , error ) ; } } ) ; channelPool . set ( offset , channelFuture ) ; } } } final CompletableFuture < Channel > future = new CompletableFuture <> ( ) ; final CompletableFuture < Channel > finalFuture = channelFuture ; finalFuture . whenComplete ( ( channel , error ) -> { if ( error == null ) { if ( ! channel . isActive ( ) ) { CompletableFuture < Channel > currentFuture ; synchronized ( channelPool ) { currentFuture = channelPool . get ( offset ) ; if ( currentFuture == finalFuture ) { channelPool . set ( offset , null ) ; } else if ( currentFuture == null ) { currentFuture = factory . apply ( address ) ; currentFuture . whenComplete ( ( c , e ) -> { if ( e == null ) { LOGGER . debug ( "Connected to {}" , channel . remoteAddress ( ) ) ; } else { LOGGER . debug ( "Failed to connect to {}" , channel . remoteAddress ( ) , e ) ; } } ) ; channelPool . set ( offset , currentFuture ) ; } } if ( currentFuture == finalFuture ) { getChannel ( address , messageType ) . whenComplete ( ( recursiveResult , recursiveError ) -> { if ( recursiveError == null ) { future . complete ( recursiveResult ) ; } else { future . completeExceptionally ( recursiveError ) ; } } ) ; } else { currentFuture . whenComplete ( ( recursiveResult , recursiveError ) -> { if ( recursiveError == null ) { future . complete ( recursiveResult ) ; } else { future . completeExceptionally ( recursiveError ) ; } } ) ; } } else { future . complete ( channel ) ; } } else { future . completeExceptionally ( error ) ; } } ) ; return future ; }
Gets or creates a pooled channel to the given address for the given message type .