docs: public static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateHttpSSE(StreamDiscovery discovery) {
          Observable<StreamAction> streamActions = discovery.getInstanceList().publish().refCount();
          Observable<StreamAction> streamAdds = streamActions.filter(a -> a.getType() == ActionType.ADD);
          Observable<StreamAction> streamRemoves = streamActions.filter(a -> a.getType() == ActionType.REMOVE);
  
          Observable<GroupedObservable<InstanceKey, Map<String, Object>>> streamPerInstance =
                  streamAdds.map(streamAction -> {
                      URI uri = streamAction.getUri();
  
                      Observable<Map<String, Object>> io = Observable.defer(() -> {
                          Observable<Map<String, Object>> flatMap = RxNetty.createHttpClient(uri.getHost(), uri.getPort(), PipelineConfigurators.<ByteBuf>sseClientConfigurator())
                                  .submit(createRequest(uri))
                                  .flatMap(response -> {
                                      if (response.getStatus().code() != 200) {
                                          return Observable.error(new RuntimeException("Failed to connect: " + response.getStatus()));
                                      }
                                      return response.getContent()
                                              .doOnSubscribe(() -> logger.info("Turbine => Aggregate Stream from URI: " + uri.toASCIIString()))
                                              .doOnUnsubscribe(() -> logger.info("Turbine => Unsubscribing Stream: " + uri))
                                              .takeUntil(streamRemoves.filter(a -> a.getUri().equals(streamAction.getUri()))) // unsubscribe when we receive a remove event
                                              .map(sse -> JsonUtility.jsonToMap(sse.getEventData()));
                                  });
                          // eclipse is having issues with type inference so breaking up 
                          return flatMap.retryWhen(attempts -> {
                              return attempts.flatMap(e -> {
                                  return Observable.timer(1, TimeUnit.SECONDS)
                                          .doOnEach(n -> logger.info("Turbine => Retrying connection to: " + uri));
                              });
                          });
  
                      });
  
                      return GroupedObservable.from(InstanceKey.create(uri.toASCIIString()), io);
                  });
  
          return StreamAggregator.aggregateGroupedStreams(streamPerInstance);
      }