code: private Observable<BucketSettings> ensureBucketIsHealthy(final Observable<BucketSettings> input) {
          return input.flatMap(new Func1<BucketSettings, Observable<BucketSettings>>() {
              @Override
              public Observable<BucketSettings> call(final BucketSettings bucketSettings) {
                  return info()
                      .delay(100, TimeUnit.MILLISECONDS)
                      .filter(new Func1<ClusterInfo, Boolean>() {
                          @Override
                          public Boolean call(ClusterInfo clusterInfo) {
                              boolean allHealthy = true;
                              for (Object n : clusterInfo.raw().getArray("nodes")) {
                                  JsonObject node = (JsonObject) n;
                                  if (!node.getString("status").equals("healthy")) {
                                      allHealthy = false;
                                      break;
                                  }
                              }
                              return allHealthy;
                          }
                      })
                      .repeat()
                      .take(1)
                      .flatMap(new Func1<ClusterInfo, Observable<BucketSettings>>() {
                          @Override
                          public Observable<BucketSettings> call(ClusterInfo clusterInfo) {
                              return Observable.just(bucketSettings);
                          }
                      });
              }
          });
      }
docs: Helper method to ensure that the state of a bucket on all nodes is healthy.
  
  This polling logic is in place as a workaround because the bucket could still be warming up or completing
  its creation process before any actual operation can be performed.
  
  @return the original input stream once done.