private Observable < Boolean > reconfigureBucket ( final BucketConfig config ) { LOGGER . debug ( "Starting reconfiguration for bucket {}" , config . name ( ) ) ; List < Observable < Boolean > > observables = new ArrayList < Observable < Boolean > > ( ) ; for ( final NodeInfo nodeInfo : config . nodes ( ) ) { final String alternate = nodeInfo . useAlternateNetwork ( ) ; final NetworkAddress altHost = alternate != null ? nodeInfo . alternateAddresses ( ) . get ( alternate ) . hostname ( ) : null ; Observable < Boolean > obs = addNode ( nodeInfo . hostname ( ) , altHost ) . flatMap ( new Func1 < LifecycleState , Observable < Map < ServiceType , Integer > > > ( ) { @ Override public Observable < Map < ServiceType , Integer > > call ( final LifecycleState lifecycleState ) { Map < ServiceType , Integer > services ; if ( alternate != null ) { AlternateAddress aa = nodeInfo . alternateAddresses ( ) . get ( alternate ) ; services = environment . sslEnabled ( ) ? aa . sslServices ( ) : aa . services ( ) ; } else { services = environment . sslEnabled ( ) ? nodeInfo . sslServices ( ) : nodeInfo . services ( ) ; } return Observable . just ( services ) ; } } ) . flatMap ( new Func1 < Map < ServiceType , Integer > , Observable < AddServiceRequest > > ( ) { @ Override public Observable < AddServiceRequest > call ( final Map < ServiceType , Integer > services ) { List < AddServiceRequest > requests = new ArrayList < AddServiceRequest > ( services . size ( ) ) ; for ( Map . Entry < ServiceType , Integer > service : services . entrySet ( ) ) { requests . add ( new AddServiceRequest ( service . getKey ( ) , config . name ( ) , config . username ( ) , config . password ( ) , service . getValue ( ) , nodeInfo . hostname ( ) ) ) ; } return Observable . from ( requests ) ; } } ) . flatMap ( new Func1 < AddServiceRequest , Observable < Service > > ( ) { @ Override public Observable < Service > call ( AddServiceRequest request ) { return addService ( request ) ; } } ) . last ( ) . map ( new Func1 < Service , Boolean > ( ) { @ Override public Boolean call ( Service service ) { return true ; } } ) ; observables . add ( obs ) ; } return Observable . merge ( observables ) . last ( ) ; }
For every bucket that is open apply the reconfiguration .