CompletableFuture < Void > identify ( ) { Member leader = context . getLeader ( ) ; if ( joinFuture != null && leader != null ) { if ( context . getLeader ( ) . equals ( member ( ) ) ) { if ( context . getState ( ) == CopycatServer . State . LEADER && ! ( ( LeaderState ) context . getServerState ( ) ) . configuring ( ) ) { if ( joinFuture != null ) joinFuture . complete ( null ) ; } else { cancelJoinTimer ( ) ; joinTimeout = context . getThreadContext ( ) . schedule ( context . getElectionTimeout ( ) . multipliedBy ( 2 ) , this :: identify ) ; } } else { cancelJoinTimer ( ) ; joinTimeout = context . getThreadContext ( ) . schedule ( context . getElectionTimeout ( ) . multipliedBy ( 2 ) , this :: identify ) ; LOGGER . debug ( "{} - Sending server identification to {}" , member ( ) . address ( ) , leader . address ( ) ) ; context . getConnections ( ) . getConnection ( leader . serverAddress ( ) ) . thenCompose ( connection -> { ReconfigureRequest request = ReconfigureRequest . builder ( ) . withIndex ( configuration . index ( ) ) . withTerm ( configuration . term ( ) ) . withMember ( member ( ) ) . build ( ) ; LOGGER . trace ( "{} - Sending {} to {}" , member . address ( ) , request , leader . address ( ) ) ; return connection . < ConfigurationRequest , ConfigurationResponse > sendAndReceive ( request ) ; } ) . whenComplete ( ( response , error ) -> { cancelJoinTimer ( ) ; if ( error == null ) { LOGGER . trace ( "{} - Received {}" , member . address ( ) , response ) ; if ( response . status ( ) == Response . Status . OK ) { if ( joinFuture != null ) { joinFuture . complete ( null ) ; } } else if ( response . error ( ) == null || response . error ( ) == CopycatError . Type . CONFIGURATION_ERROR ) { LOGGER . debug ( "{} - Failed to update configuration: configuration change already in progress" , member . address ( ) ) ; joinTimeout = context . getThreadContext ( ) . schedule ( context . getElectionTimeout ( ) . multipliedBy ( 2 ) , this :: identify ) ; } } else { LOGGER . warn ( "{} - Failed to update configuration: {}" , member . address ( ) , error . getMessage ( ) ) ; joinTimeout = context . getThreadContext ( ) . schedule ( context . getElectionTimeout ( ) . multipliedBy ( 2 ) , this :: identify ) ; } } ) ; } } return joinFuture ; }
Identifies this server to the leader .