FlatMap for stream aggregation