@ 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 ) ; } ) ; }
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 .