@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);
            });
    }
@ 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 -> { 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 ( ) ) ; } ) . scan ( new HashMap < String , Object > ( ) , StreamAggregator :: sumOfDelta ) . skip ( 1 ) ; return GroupedObservable . from ( commandGroup . getKey ( ) , sumOfDeltasForAllInstancesForCommand ) ; } ) ; }
Flatten the stream and then do nested groupBy. This matches the mental model and is simple but
serializes the stream to a single thread which is bad for performance.

TODO: Add a version like this but with a ParallelObservable

@param stream
@return