docs: @SuppressWarnings("unused")
      private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingFlattenedGroupBy(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> stream) {
          Observable<Map<String, Object>> allData = stream.flatMap(instanceStream -> {
              return instanceStream
                      .map((Map<String, Object> event) -> {
                          event.put("InstanceKey", instanceStream.getKey());
                          event.put("TypeAndName", TypeAndNameKey.from(String.valueOf(event.get("type")), String.valueOf(event.get("name"))));
                          return event;
                      })
                      .compose(is -> {
                          return tombstone(is, instanceStream.getKey());
                      });
          });
  
          Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> byCommand = allData
                  .groupBy((Map<String, Object> event) -> {
                      return (TypeAndNameKey) event.get("TypeAndName");
                  });
  
          return byCommand
                  .map(commandGroup -> {
                      Observable<Map<String, Object>> sumOfDeltasForAllInstancesForCommand = commandGroup
                              .groupBy((Map<String, Object> json) -> {
                                  return json.get("InstanceKey");
                              }).flatMap(instanceGroup -> {
                                  // calculate and output deltas for each instance stream per command
                                      return instanceGroup
                                              .takeUntil(d -> d.containsKey("tombstone"))
                                              .startWith(Collections.<String, Object> emptyMap())
                                              .map(data -> {
                                                  if (data.containsKey("tombstone")) {
                                                      return Collections.<String, Object> emptyMap();
                                                  } else {
                                                      return data;
                                                  }
                                              })
                                              .buffer(2, 1)
                                              .filter(list -> list.size() == 2)
                                              .map(StreamAggregator::previousAndCurrentToDelta)
                                              .filter(data -> data != null && !data.isEmpty());
                                  })
                              // we now have all instance deltas merged into a single stream per command
                              // and sum them into a single stream of aggregate values
                              .scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta)
                              .skip(1);
  
                      // we artificially wrap in a GroupedObservable that communicates the CommandKey this stream represents
                  return GroupedObservable.from(commandGroup.getKey(), sumOfDeltasForAllInstancesForCommand);
              });
      }