fix(agent): carry a parallel pipeline through sorted, limit, parallel() and the other lambda-free stages (#1172)

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

@@ -53,6 +53,12 @@
 import java.util.stream.LongStream;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
+import java.util.function.IntToLongFunction;
+import java.util.function.IntToDoubleFunction;
+import java.util.function.LongToIntFunction;
+import java.util.function.LongToDoubleFunction;
+import java.util.function.DoubleToIntFunction;
+import java.util.function.DoubleToLongFunction;
 import java.util.function.IntUnaryOperator;
 import java.util.function.IntPredicate;
 import java.util.function.IntBinaryOperator;
@@ -904,6 +910,55 @@ static Set<String> wovenCallSites() {
             Entry.call(Stream.class, "count", "streamCount"),
             Entry.call(Stream.class, "findFirst", "streamFindFirst"),
             Entry.call(Stream.class, "findAny", "streamFindAny"),
+            // Stages that run no lambda carry the woven stages behind them, and the remaining
+            // lambda stages wait in slots as map does (#1172).
+            Entry.call(Stream.class, "sorted", "streamSorted"),
+            Entry.call(Stream.class, "distinct", "streamDistinct"),
+            Entry.call(Stream.class, "limit", "streamLimit", long.class),
+            Entry.call(Stream.class, "skip", "streamSkip", long.class),
+            Entry.call(Stream.class, "parallel", "streamParallel"),
+            Entry.call(Stream.class, "sequential", "streamSequential"),
+            Entry.call(Stream.class, "unordered", "streamUnordered"),
+            Entry.call(Stream.class, "sorted", "streamSorted", Comparator.class),
+            Entry.call(Stream.class, "flatMap", "streamFlatMap", Function.class),
+            Entry.call(Stream.class, "takeWhile", "streamTakeWhile", Predicate.class),
+            Entry.call(Stream.class, "dropWhile", "streamDropWhile", Predicate.class),
+            Entry.call(IntStream.class, "sorted", "intStreamSorted"),
+            Entry.call(IntStream.class, "distinct", "intStreamDistinct"),
+            Entry.call(IntStream.class, "limit", "intStreamLimit", long.class),
+            Entry.call(IntStream.class, "skip", "intStreamSkip", long.class),
+            Entry.call(IntStream.class, "parallel", "intStreamParallel"),
+            Entry.call(IntStream.class, "sequential", "intStreamSequential"),
+            Entry.call(IntStream.class, "flatMap", "intStreamFlatMap", IntFunction.class),
+            Entry.call(IntStream.class, "takeWhile", "intStreamTakeWhile", IntPredicate.class),
+            Entry.call(IntStream.class, "dropWhile", "intStreamDropWhile", IntPredicate.class),
+            Entry.call(IntStream.class, "mapToLong", "intStreamMapToLong", IntToLongFunction.class),
+            Entry.call(IntStream.class, "mapToDouble", "intStreamMapToDouble", IntToDoubleFunction.class),
+            Entry.call(IntStream.class, "asLongStream", "intStreamAsLongStream"),
+            Entry.call(IntStream.class, "asDoubleStream", "intStreamAsDoubleStream"),
+            Entry.call(LongStream.class, "sorted", "longStreamSorted"),
+            Entry.call(LongStream.class, "distinct", "longStreamDistinct"),
+            Entry.call(LongStream.class, "limit", "longStreamLimit", long.class),
+            Entry.call(LongStream.class, "skip", "longStreamSkip", long.class),
+            Entry.call(LongStream.class, "parallel", "longStreamParallel"),
+            Entry.call(LongStream.class, "sequential", "longStreamSequential"),
+            Entry.call(LongStream.class, "flatMap", "longStreamFlatMap", LongFunction.class),
+            Entry.call(LongStream.class, "takeWhile", "longStreamTakeWhile", LongPredicate.class),
+            Entry.call(LongStream.class, "dropWhile", "longStreamDropWhile", LongPredicate.class),
+            Entry.call(LongStream.class, "mapToInt", "longStreamMapToInt", LongToIntFunction.class),
+            Entry.call(LongStream.class, "mapToDouble", "longStreamMapToDouble", LongToDoubleFunction.class),
+            Entry.call(LongStream.class, "asDoubleStream", "longStreamAsDoubleStream"),
+            Entry.call(DoubleStream.class, "sorted", "doubleStreamSorted"),
+            Entry.call(DoubleStream.class, "distinct", "doubleStreamDistinct"),
+            Entry.call(DoubleStream.class, "limit", "doubleStreamLimit", long.class),
+            Entry.call(DoubleStream.class, "skip", "doubleStreamSkip", long.class),
+            Entry.call(DoubleStream.class, "parallel", "doubleStreamParallel"),
+            Entry.call(DoubleStream.class, "sequential", "doubleStreamSequential"),
+            Entry.call(DoubleStream.class, "flatMap", "doubleStreamFlatMap", DoubleFunction.class),
+            Entry.call(DoubleStream.class, "takeWhile", "doubleStreamTakeWhile", DoublePredicate.class),
+            Entry.call(DoubleStream.class, "dropWhile", "doubleStreamDropWhile", DoublePredicate.class),
+            Entry.call(DoubleStream.class, "mapToInt", "doubleStreamMapToInt", DoubleToIntFunction.class),
+            Entry.call(DoubleStream.class, "mapToLong", "doubleStreamMapToLong", DoubleToLongFunction.class),
             // The same for IntStream, LongStream and DoubleStream, and across the bridges
             // between a Stream and them: mapToInt, mapToLong, mapToDouble, mapToObj and boxed
             // keep the stages behind them in the pipeline they return (#1172).

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

@@ -774,6 +774,194 @@ public static java.util.Optional<Object> streamFindAny(java.util.stream.Stream<O
             return receiver.findAny();
         }
 
+        public static java.util.stream.Stream<Object> streamSorted(java.util.stream.Stream<Object> receiver) {
+            return receiver.sorted();
+        }
+
+        public static java.util.stream.Stream<Object> streamDistinct(java.util.stream.Stream<Object> receiver) {
+            return receiver.distinct();
+        }
+
+        public static java.util.stream.Stream<Object> streamLimit(java.util.stream.Stream<Object> receiver, long maxSize) {
+            return receiver.limit(maxSize);
+        }
+
+        public static java.util.stream.Stream<Object> streamSkip(java.util.stream.Stream<Object> receiver, long n) {
+            return receiver.skip(n);
+        }
+
+        public static java.util.stream.BaseStream<?, ?> streamParallel(java.util.stream.Stream<Object> receiver) {
+            return receiver.parallel();
+        }
+
+        public static java.util.stream.BaseStream<?, ?> streamSequential(java.util.stream.Stream<Object> receiver) {
+            return receiver.sequential();
+        }
+
+        public static java.util.stream.BaseStream<?, ?> streamUnordered(java.util.stream.Stream<Object> receiver) {
+            return receiver.unordered();
+        }
+
+        public static java.util.stream.Stream<Object> streamSorted(java.util.stream.Stream<Object> receiver, java.util.Comparator<Object> comparator) {
+            return receiver.sorted(comparator);
+        }
+
+        public static java.util.stream.Stream<Object> streamFlatMap(java.util.stream.Stream<Object> receiver, java.util.function.Function<Object, java.util.stream.Stream<Object>> mapper) {
+            return receiver.flatMap(mapper);
+        }
+
+        public static java.util.stream.Stream<Object> streamTakeWhile(java.util.stream.Stream<Object> receiver, java.util.function.Predicate<Object> predicate) {
+            return receiver.takeWhile(predicate);
+        }
+
+        public static java.util.stream.Stream<Object> streamDropWhile(java.util.stream.Stream<Object> receiver, java.util.function.Predicate<Object> predicate) {
+            return receiver.dropWhile(predicate);
+        }
+
+        public static java.util.stream.IntStream intStreamSorted(java.util.stream.IntStream receiver) {
+            return receiver.sorted();
+        }
+
+        public static java.util.stream.IntStream intStreamDistinct(java.util.stream.IntStream receiver) {
+            return receiver.distinct();
+        }
+
+        public static java.util.stream.IntStream intStreamLimit(java.util.stream.IntStream receiver, long maxSize) {
+            return receiver.limit(maxSize);
+        }
+
+        public static java.util.stream.IntStream intStreamSkip(java.util.stream.IntStream receiver, long n) {
+            return receiver.skip(n);
+        }
+
+        public static java.util.stream.IntStream intStreamParallel(java.util.stream.IntStream receiver) {
+            return receiver.parallel();
+        }
+
+        public static java.util.stream.IntStream intStreamSequential(java.util.stream.IntStream receiver) {
+            return receiver.sequential();
+        }
+
+        public static java.util.stream.IntStream intStreamFlatMap(java.util.stream.IntStream receiver, java.util.function.IntFunction<java.util.stream.IntStream> mapper) {
+            return receiver.flatMap(mapper);
+        }
+
+        public static java.util.stream.IntStream intStreamTakeWhile(java.util.stream.IntStream receiver, java.util.function.IntPredicate predicate) {
+            return receiver.takeWhile(predicate);
+        }
+
+        public static java.util.stream.IntStream intStreamDropWhile(java.util.stream.IntStream receiver, java.util.function.IntPredicate predicate) {
+            return receiver.dropWhile(predicate);
+        }
+
+        public static java.util.stream.LongStream intStreamMapToLong(java.util.stream.IntStream receiver, java.util.function.IntToLongFunction mapper) {
+            return receiver.mapToLong(mapper);
+        }
+
+        public static java.util.stream.DoubleStream intStreamMapToDouble(java.util.stream.IntStream receiver, java.util.function.IntToDoubleFunction mapper) {
+            return receiver.mapToDouble(mapper);
+        }
+
+        public static java.util.stream.LongStream intStreamAsLongStream(java.util.stream.IntStream receiver) {
+            return receiver.asLongStream();
+        }
+
+        public static java.util.stream.DoubleStream intStreamAsDoubleStream(java.util.stream.IntStream receiver) {
+            return receiver.asDoubleStream();
+        }
+
+        public static java.util.stream.LongStream longStreamSorted(java.util.stream.LongStream receiver) {
+            return receiver.sorted();
+        }
+
+        public static java.util.stream.LongStream longStreamDistinct(java.util.stream.LongStream receiver) {
+            return receiver.distinct();
+        }
+
+        public static java.util.stream.LongStream longStreamLimit(java.util.stream.LongStream receiver, long maxSize) {
+            return receiver.limit(maxSize);
+        }
+
+        public static java.util.stream.LongStream longStreamSkip(java.util.stream.LongStream receiver, long n) {
+            return receiver.skip(n);
+        }
+
+        public static java.util.stream.LongStream longStreamParallel(java.util.stream.LongStream receiver) {
+            return receiver.parallel();
+        }
+
+        public static java.util.stream.LongStream longStreamSequential(java.util.stream.LongStream receiver) {
+            return receiver.sequential();
+        }
+
+        public static java.util.stream.LongStream longStreamFlatMap(java.util.stream.LongStream receiver, java.util.function.LongFunction<java.util.stream.LongStream> mapper) {
+            return receiver.flatMap(mapper);
+        }
+
+        public static java.util.stream.LongStream longStreamTakeWhile(java.util.stream.LongStream receiver, java.util.function.LongPredicate predicate) {
+            return receiver.takeWhile(predicate);
+        }
+
+        public static java.util.stream.LongStream longStreamDropWhile(java.util.stream.LongStream receiver, java.util.function.LongPredicate predicate) {
+            return receiver.dropWhile(predicate);
+        }
+
+        public static java.util.stream.IntStream longStreamMapToInt(java.util.stream.LongStream receiver, java.util.function.LongToIntFunction mapper) {
+            return receiver.mapToInt(mapper);
+        }
+
+        public static java.util.stream.DoubleStream longStreamMapToDouble(java.util.stream.LongStream receiver, java.util.function.LongToDoubleFunction mapper) {
+            return receiver.mapToDouble(mapper);
+        }
+
+        public static java.util.stream.DoubleStream longStreamAsDoubleStream(java.util.stream.LongStream receiver) {
+            return receiver.asDoubleStream();
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamSorted(java.util.stream.DoubleStream receiver) {
+            return receiver.sorted();
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamDistinct(java.util.stream.DoubleStream receiver) {
+            return receiver.distinct();
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamLimit(java.util.stream.DoubleStream receiver, long maxSize) {
+            return receiver.limit(maxSize);
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamSkip(java.util.stream.DoubleStream receiver, long n) {
+            return receiver.skip(n);
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamParallel(java.util.stream.DoubleStream receiver) {
+            return receiver.parallel();
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamSequential(java.util.stream.DoubleStream receiver) {
+            return receiver.sequential();
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamFlatMap(java.util.stream.DoubleStream receiver, java.util.function.DoubleFunction<java.util.stream.DoubleStream> mapper) {
+            return receiver.flatMap(mapper);
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamTakeWhile(java.util.stream.DoubleStream receiver, java.util.function.DoublePredicate predicate) {
+            return receiver.takeWhile(predicate);
+        }
+
+        public static java.util.stream.DoubleStream doubleStreamDropWhile(java.util.stream.DoubleStream receiver, java.util.function.DoublePredicate predicate) {
+            return receiver.dropWhile(predicate);
+        }
+
+        public static java.util.stream.IntStream doubleStreamMapToInt(java.util.stream.DoubleStream receiver, java.util.function.DoubleToIntFunction mapper) {
+            return receiver.mapToInt(mapper);
+        }
+
+        public static java.util.stream.LongStream doubleStreamMapToLong(java.util.stream.DoubleStream receiver, java.util.function.DoubleToLongFunction mapper) {
+            return receiver.mapToLong(mapper);
+        }
+
         public static java.util.stream.IntStream streamMapToInt(java.util.stream.Stream<Object> receiver, java.util.function.ToIntFunction<Object> mapper) {
             return receiver.mapToInt(mapper);
         }