private Intermediate mapAndReduce ( Context ctx ) { /* Fetch each metric wildcard and add it to the to-be-processed list. */ final List < Map . Entry < Tags , MetricValue > > matcher_tsvs = matchers_ . stream ( ) . flatMap ( m -> m . filter ( ctx ) ) . map ( named_entry -> SimpleMapEntry . create ( named_entry . getKey ( ) . getTags ( ) , named_entry . getValue ( ) ) ) . collect ( Collectors . toList ( ) ) ; /* Resolve each expression. */ final Map < Boolean , List < TimeSeriesMetricDeltaSet > > expr_tsvs_map = exprs_ . stream ( ) . map ( expr -> expr . apply ( ctx ) ) . collect ( Collectors . partitioningBy ( TimeSeriesMetricDeltaSet :: isVector ) ) ; final List < TimeSeriesMetricDeltaSet > expr_tsvs = expr_tsvs_map . getOrDefault ( true , Collections . emptyList ( ) ) ; final Optional < T > scalar = expr_tsvs_map . getOrDefault ( false , Collections . emptyList ( ) ) . stream ( ) . map ( tsv_set -> map_ ( tsv_set . asScalar ( ) . get ( ) ) ) . reduce ( this :: reducer_ ) ; /* * Reduce everything using the reducer (in the derived class). */ return new Intermediate ( scalar , unmodifiableMap ( aggregation_ . apply ( matcher_tsvs . stream ( ) , expr_tsvs . stream ( ) . flatMap ( TimeSeriesMetricDeltaSet :: streamAsMap ) , Map . Entry :: getKey , Map . Entry :: getKey , Map . Entry :: getValue , Map . Entry :: getValue ) . entrySet ( ) . stream ( ) . map ( entry -> { return entry . getValue ( ) . stream ( ) . map ( this :: map_ ) . reduce ( this :: reducer_ ) . map ( v -> SimpleMapEntry . create ( entry . getKey ( ) , v ) ) ; } ) . flatMap ( opt -> opt . map ( Stream :: of ) . orElseGet ( Stream :: empty ) ) . collect ( Collectors . toMap ( entry -> entry . getKey ( ) , entry -> entry . getValue ( ) ) ) ) ) ; }
Create a reduction of the matchers and expressions for a given context .