code: private void repeatConfigUntilUnsubscribed(final String name, Observable<BucketStreamingResponse> response) {
          response.flatMap(new Func1<BucketStreamingResponse, Observable<ProposedBucketConfigContext>>() {
              @Override
              public Observable<ProposedBucketConfigContext> call(final BucketStreamingResponse response) {
                  LOGGER.debug("Config stream started for {} on {}.", name, response.host());
  
                  return response
                      .configs()
                      .map(new Func1<String, ProposedBucketConfigContext>() {
                          @Override
                          public ProposedBucketConfigContext call(String s) {
                              String raw = s.replace("$HOST", response.host());
                              return new ProposedBucketConfigContext(name, raw, NetworkAddress.create(response.host()));
                          }
                      })
                      .doOnCompleted(new Action0() {
                          @Override
                          public void call() {
                              LOGGER.debug("Config stream ended for {} on {}.", name, response.host());
                          }
                      });
              }
          })
          .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
              @Override
              public Observable<?> call(Observable<? extends Void> observable) {
                  return observable.flatMap(new Func1<Void, Observable<?>>() {
                      @Override
                      public Observable<?> call(Void aVoid) {
                          if (HttpRefresher.this.registrations().containsKey(name)) {
                              LOGGER.debug("Resubscribing config stream for bucket {}, still registered.", name);
                              return Observable.just(true);
                          } else {
                              LOGGER.debug("Not resubscribing config stream for bucket {}, not registered.", name);
                              return Observable.empty();
                          }
                      }
                  });
              }
          }).subscribe(new Subscriber<ProposedBucketConfigContext>() {
              @Override
              public void onCompleted() {
                  // ignored on purpose
              }
  
              @Override
              public void onError(Throwable e) {
                  LOGGER.error("Error while subscribing to Http refresh stream!", e);
              }
  
              @Override
              public void onNext(ProposedBucketConfigContext ctx) {
                  pushConfig(ctx);
              }
          });
      }
docs: Helper method to push configs until unsubscribed, even when a stream closes.
  
  @param name the name of the bucket.
  @param response the response source observable to resubscribe if needed.