private static Observable < Map < String , Object > > tombstone ( Observable < Map < String , Object > > instanceStream , InstanceKey instanceKey ) { return instanceStream . publish ( is -> { Observable < Map < String , Object > > tombstone = is // collect all unique "TypeAndName" keys . collect ( ( ) -> new HashSet < TypeAndNameKey > ( ) , ( listOfTypeAndName , event ) -> { listOfTypeAndName . add ( ( TypeAndNameKey ) event . get ( "TypeAndName" ) ) ; } ) // when instanceStream completes, emit a "tombstone" for each "TypeAndName" in the HashSet collected above . flatMap ( listOfTypeAndName -> { return Observable . from ( listOfTypeAndName ) . map ( typeAndName -> { Map < String , Object > tombstoneForTypeAndName = new LinkedHashMap <> ( ) ; tombstoneForTypeAndName . put ( "tombstone" , "true" ) ; tombstoneForTypeAndName . put ( "InstanceKey" , instanceKey ) ; tombstoneForTypeAndName . put ( "TypeAndName" , typeAndName ) ; tombstoneForTypeAndName . put ( "name" , typeAndName . getName ( ) ) ; tombstoneForTypeAndName . put ( "type" , typeAndName . getType ( ) ) ; return tombstoneForTypeAndName ; } ) ; } ) ; // concat the tombstone events to the end of the original stream return is . mergeWith ( tombstone ) ; } ) ; }
Append tombstone events to each instanceStream when they terminate so that the last values from the stream will be removed from all aggregates down stream .