@ SuppressWarnings ( "unchecked" ) private void register ( final G gateway , final int attempt , final long timeoutMillis ) { // eager check for canceling to avoid some unnecessary work if ( canceled ) { return ; } try { log . info ( "Registration at {} attempt {} (timeout={}ms)" , targetName , attempt , timeoutMillis ) ; CompletableFuture < RegistrationResponse > registrationFuture = invokeRegistration ( gateway , fencingToken , timeoutMillis ) ; // if the registration was successful, let the TaskExecutor know CompletableFuture < Void > registrationAcceptFuture = registrationFuture . thenAcceptAsync ( ( RegistrationResponse result ) -> { if ( ! isCanceled ( ) ) { if ( result instanceof RegistrationResponse . Success ) { // registration successful! S success = ( S ) result ; completionFuture . complete ( Tuple2 . of ( gateway , success ) ) ; } else { // registration refused or unknown if ( result instanceof RegistrationResponse . Decline ) { RegistrationResponse . Decline decline = ( RegistrationResponse . Decline ) result ; log . info ( "Registration at {} was declined: {}" , targetName , decline . getReason ( ) ) ; } else { log . error ( "Received unknown response to registration attempt: {}" , result ) ; } log . info ( "Pausing and re-attempting registration in {} ms" , retryingRegistrationConfiguration . getRefusedDelayMillis ( ) ) ; registerLater ( gateway , 1 , retryingRegistrationConfiguration . getInitialRegistrationTimeoutMillis ( ) , retryingRegistrationConfiguration . getRefusedDelayMillis ( ) ) ; } } } , rpcService . getExecutor ( ) ) ; // upon failure, retry registrationAcceptFuture . whenCompleteAsync ( ( Void v , Throwable failure ) -> { if ( failure != null && ! isCanceled ( ) ) { if ( ExceptionUtils . stripCompletionException ( failure ) instanceof TimeoutException ) { // we simply have not received a response in time. maybe the timeout was // very low (initial fast registration attempts), maybe the target endpoint is // currently down. if ( log . isDebugEnabled ( ) ) { log . debug ( "Registration at {} ({}) attempt {} timed out after {} ms" , targetName , targetAddress , attempt , timeoutMillis ) ; } long newTimeoutMillis = Math . min ( 2 * timeoutMillis , retryingRegistrationConfiguration . getMaxRegistrationTimeoutMillis ( ) ) ; register ( gateway , attempt + 1 , newTimeoutMillis ) ; } else { // a serious failure occurred. we still should not give up, but keep trying log . error ( "Registration at {} failed due to an error" , targetName , failure ) ; log . info ( "Pausing and re-attempting registration in {} ms" , retryingRegistrationConfiguration . getErrorDelayMillis ( ) ) ; registerLater ( gateway , 1 , retryingRegistrationConfiguration . getInitialRegistrationTimeoutMillis ( ) , retryingRegistrationConfiguration . getErrorDelayMillis ( ) ) ; } } } , rpcService . getExecutor ( ) ) ; } catch ( Throwable t ) { completionFuture . completeExceptionally ( t ) ; cancel ( ) ; } }
This method performs a registration attempt and triggers either a success notification or a retry depending on the result .