fix(agent): follow a parallel Stream's map, filter, reduce and collect lambdas (#1172)

async-test-agent/src/main/java/se/deversity/asynctest/agent/CollectionAccessWeaver.java

@@ -34,6 +34,10 @@
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.IntFunction;
+import java.util.Comparator;
+import java.util.function.BinaryOperator;
+import java.util.stream.Collector;
 import java.util.function.BiFunction;
 import java.util.function.BiConsumer;
 import java.util.concurrent.CompletionStage;
@@ -857,6 +861,32 @@ static Set<String> wovenCallSites() {
             Entry.call(IntStream.class, "forEach", "intStreamForEach", IntConsumer.class),
             Entry.call(LongStream.class, "forEach", "longStreamForEach", LongConsumer.class),
             Entry.call(DoubleStream.class, "forEach", "doubleStreamForEach", DoubleConsumer.class),
+            // The rest of a Stream pipeline (#1172): a lambda given to map, filter or peek waits in
+            // a slot the terminal operation lends the run to while it runs, and a terminal
+            // operation's own lambdas run like forEach's action. A terminal operation not listed
+            // here lends nothing, so its pipeline runs unattributed as before.
+            Entry.call(Stream.class, "map", "streamMap", Function.class),
+            Entry.call(Stream.class, "filter", "streamFilter", Predicate.class),
+            Entry.call(Stream.class, "peek", "streamPeek", Consumer.class),
+            Entry.call(Stream.class, "forEachOrdered", "streamForEachOrdered", Consumer.class),
+            Entry.call(Stream.class, "collect", "streamCollect", Collector.class),
+            Entry.call(Stream.class, "collect", "streamCollect", Supplier.class, BiConsumer.class,
+                    BiConsumer.class),
+            Entry.call(Stream.class, "reduce", "streamReduce", BinaryOperator.class),
+            Entry.call(Stream.class, "reduce", "streamReduce", Object.class, BinaryOperator.class),
+            Entry.call(Stream.class, "reduce", "streamReduce", Object.class, BiFunction.class,
+                    BinaryOperator.class),
+            Entry.call(Stream.class, "min", "streamMin", Comparator.class),
+            Entry.call(Stream.class, "max", "streamMax", Comparator.class),
+            Entry.call(Stream.class, "anyMatch", "streamAnyMatch", Predicate.class),
+            Entry.call(Stream.class, "allMatch", "streamAllMatch", Predicate.class),
+            Entry.call(Stream.class, "noneMatch", "streamNoneMatch", Predicate.class),
+            Entry.call(Stream.class, "toList", "streamToList"),
+            Entry.call(Stream.class, "toArray", "streamToArray"),
+            Entry.call(Stream.class, "toArray", "streamToArray", IntFunction.class),
+            Entry.call(Stream.class, "count", "streamCount"),
+            Entry.call(Stream.class, "findFirst", "streamFindFirst"),
+            Entry.call(Stream.class, "findAny", "streamFindAny"),
             // A submit whose future the caller pops can never be inspected, so the task's
             // exception reaches nobody; the POP becomes submitResultDiscarded, and a store, a
             // return or a get is left alone.

async-test-agent/src/test/java/se/deversity/asynctest/agent/DiscardedOfferWeavingTest.java

@@ -685,6 +685,95 @@ public static void doubleStreamForEach(java.util.stream.DoubleStream receiver,
             receiver.forEach(action);
         }
 
+        public static java.util.stream.Stream<Object> streamMap(java.util.stream.Stream<Object> receiver, java.util.function.Function<Object, Object> mapper) {
+            return receiver.map(mapper);
+        }
+
+        public static java.util.stream.Stream<Object> streamFilter(java.util.stream.Stream<Object> receiver, java.util.function.Predicate<Object> predicate) {
+            return receiver.filter(predicate);
+        }
+
+        public static java.util.stream.Stream<Object> streamPeek(java.util.stream.Stream<Object> receiver, java.util.function.Consumer<Object> action) {
+            return receiver.peek(action);
+        }
+
+        public static void streamForEachOrdered(java.util.stream.Stream<Object> receiver, java.util.function.Consumer<Object> action) {
+            receiver.forEachOrdered(action);
+        }
+
+        public static Object streamCollect(java.util.stream.Stream<Object> receiver,
+                                           java.util.stream.Collector<? super Object, ?, ?> collector) {
+            return receiver.collect(collector);
+        }
+
+        public static Object streamCollect(java.util.stream.Stream<Object> receiver, java.util.function.Supplier<Object> supplier,
+                                           java.util.function.BiConsumer<Object, Object> accumulator,
+                                           java.util.function.BiConsumer<Object, Object> combiner) {
+            return receiver.collect(supplier, accumulator, combiner);
+        }
+
+        public static java.util.Optional<Object> streamReduce(java.util.stream.Stream<Object> receiver,
+                                                              java.util.function.BinaryOperator<Object> accumulator) {
+            return receiver.reduce(accumulator);
+        }
+
+        public static Object streamReduce(java.util.stream.Stream<Object> receiver, Object identity,
+                                          java.util.function.BinaryOperator<Object> accumulator) {
+            return receiver.reduce(identity, accumulator);
+        }
+
+        public static Object streamReduce(java.util.stream.Stream<Object> receiver, Object identity,
+                                          java.util.function.BiFunction<Object, Object, Object> accumulator,
+                                          java.util.function.BinaryOperator<Object> combiner) {
+            return receiver.reduce(identity, accumulator, combiner);
+        }
+
+        public static java.util.Optional<Object> streamMin(java.util.stream.Stream<Object> receiver,
+                                                           java.util.Comparator<Object> comparator) {
+            return receiver.min(comparator);
+        }
+
+        public static java.util.Optional<Object> streamMax(java.util.stream.Stream<Object> receiver,
+                                                           java.util.Comparator<Object> comparator) {
+            return receiver.max(comparator);
+        }
+
+        public static boolean streamAnyMatch(java.util.stream.Stream<Object> receiver, java.util.function.Predicate<Object> predicate) {
+            return receiver.anyMatch(predicate);
+        }
+
+        public static boolean streamAllMatch(java.util.stream.Stream<Object> receiver, java.util.function.Predicate<Object> predicate) {
+            return receiver.allMatch(predicate);
+        }
+
+        public static boolean streamNoneMatch(java.util.stream.Stream<Object> receiver, java.util.function.Predicate<Object> predicate) {
+            return receiver.noneMatch(predicate);
+        }
+
+        public static java.util.List<Object> streamToList(java.util.stream.Stream<Object> receiver) {
+            return receiver.toList();
+        }
+
+        public static Object[] streamToArray(java.util.stream.Stream<Object> receiver) {
+            return receiver.toArray();
+        }
+
+        public static Object[] streamToArray(java.util.stream.Stream<Object> receiver, java.util.function.IntFunction<Object[]> generator) {
+            return receiver.toArray(generator);
+        }
+
+        public static long streamCount(java.util.stream.Stream<Object> receiver) {
+            return receiver.count();
+        }
+
+        public static java.util.Optional<Object> streamFindFirst(java.util.stream.Stream<Object> receiver) {
+            return receiver.findFirst();
+        }
+
+        public static java.util.Optional<Object> streamFindAny(java.util.stream.Stream<Object> receiver) {
+            return receiver.findAny();
+        }
+
         public static java.util.concurrent.CompletableFuture<Void> runAsync(Runnable r) {
             return java.util.concurrent.CompletableFuture.runAsync(r);
         }