From 40b67f12102e67888d01c83ac4b8b70f59c2748f Mon Sep 17 00:00:00 2001 From: akarnokd Date: Sat, 18 Jul 2026 11:39:03 +0200 Subject: [PATCH 1/2] 4.x: Streamable Operator optimizations 7/18 --- .../streamable/StreamableCollectPerf.java | 14 ++ .../streamable/StreamableFilterPerf.java | 177 +++++++++++++++ .../StreamableLastAsSinglePerf.java | 15 ++ .../streamable/StreamableCollector.java | 71 +++--- .../operators/streamable/StreamableEmpty.java | 22 +- .../streamable/StreamableFilter.java | 207 +++++++++++++++--- .../operators/streamable/StreamableJust.java | 27 ++- .../streamable/StreamableLastAsSingle.java | 66 +++--- .../operators/streamable/StreamableMap.java | 17 +- .../operators/streamable/StreamableRange.java | 13 +- .../streamable/StreamableRangeLong.java | 25 ++- .../streamable/StreamableCollectorTest.java | 54 +++++ .../StreamableConcatIterableTest.java | 16 ++ .../streamable/StreamableEmptyTest.java | 67 ++++++ .../streamable/StreamableFilterTest.java | 195 ++++++++++++++++- .../streamable/StreamableJustTest.java | 53 +++++ .../StreamableLastAsSingleTest.java | 10 + .../streamable/StreamableMapTest.java | 10 + .../streamable/StreamableRangeTest.java | 22 ++ 19 files changed, 974 insertions(+), 107 deletions(-) create mode 100644 src/jmh/java/io/reactivex/rxjava4/streamable/StreamableFilterPerf.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJustTest.java diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableCollectPerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableCollectPerf.java index 8ccfb19a16..5ec4a0a880 100644 --- a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableCollectPerf.java +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableCollectPerf.java @@ -102,6 +102,20 @@ /// StreamableCollectPerf.benchmarkHidden 100000 thrpt 5 1161,570 ┬▒ 20,902 ops/s /// StreamableCollectPerf.benchmarkHidden 1000000 thrpt 5 116,332 ┬▒ 1,313 ops/s /// ``` +/// +/// ## 5 - Synchronous bias in the non-fused area, via Claude Fable +/// +/// +198% for times 1 vs 4., +177% for times million vs 4 +/// +/// ``` +/// StreamableCollectPerf.benchmarkHidden 1 thrpt 5 47849604,612 ┬▒ 873982,158 ops/s +/// StreamableCollectPerf.benchmarkHidden 10 thrpt 5 34146288,027 ┬▒ 395418,081 ops/s +/// StreamableCollectPerf.benchmarkHidden 100 thrpt 5 3294354,815 ┬▒ 47802,050 ops/s +/// StreamableCollectPerf.benchmarkHidden 1000 thrpt 5 329341,300 ┬▒ 6665,423 ops/s +/// StreamableCollectPerf.benchmarkHidden 10000 thrpt 5 33513,984 ┬▒ 2196,936 ops/s +/// StreamableCollectPerf.benchmarkHidden 100000 thrpt 5 3529,826 ┬▒ 71,479 ops/s +/// StreamableCollectPerf.benchmarkHidden 1000000 thrpt 5 322,838 ┬▒ 5,698 ops/s +/// ``` @BenchmarkMode(Mode.Throughput) @Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) @Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableFilterPerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableFilterPerf.java new file mode 100644 index 0000000000..aa55ca97cd --- /dev/null +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableFilterPerf.java @@ -0,0 +1,177 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.streamable; + +import java.util.*; +import java.util.concurrent.TimeUnit; +import java.util.stream.*; + +import org.openjdk.jmh.annotations.*; + +import io.reactivex.rxjava4.core.Streamable; + +/// +/// The map is one of the most used operator in the ecosystem so it must be fast and +/// it must support operator fusion across itself. +/// +/// i9 275HX, 32GB LPDDR5 6400MT CL52, Windows 25H2, JDK 26.0.1 +/// +/// # 0. Baseline +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableFilterPerf.basic 1 thrpt 5 26170031,056 ┬▒ 100892,180 ops/s +/// StreamableFilterPerf.basic 10 thrpt 5 4174832,259 ┬▒ 118915,456 ops/s +/// StreamableFilterPerf.basic 100 thrpt 5 485563,469 ┬▒ 6977,640 ops/s +/// StreamableFilterPerf.basic 1000 thrpt 5 52859,526 ┬▒ 392,085 ops/s +/// StreamableFilterPerf.basic 10000 thrpt 5 4757,976 ┬▒ 67,715 ops/s +/// StreamableFilterPerf.basic 100000 thrpt 5 475,093 ┬▒ 6,846 ops/s +/// StreamableFilterPerf.basic 1000000 thrpt 5 47,684 ┬▒ 1,866 ops/s +/// StreamableFilterPerf.enumerated 1 thrpt 5 9351050,757 ┬▒ 267315,421 ops/s +/// StreamableFilterPerf.enumerated 10 thrpt 5 3315310,614 ┬▒ 72388,028 ops/s +/// StreamableFilterPerf.enumerated 100 thrpt 5 377616,877 ┬▒ 8096,437 ops/s +/// StreamableFilterPerf.enumerated 1000 thrpt 5 41714,551 ┬▒ 813,698 ops/s +/// StreamableFilterPerf.enumerated 10000 thrpt 5 4084,944 ┬▒ 64,690 ops/s +/// StreamableFilterPerf.enumerated 100000 thrpt 5 367,438 ┬▒ 10,352 ops/s +/// StreamableFilterPerf.enumerated 1000000 thrpt 5 37,380 ┬▒ 0,771 ops/s +/// StreamableFilterPerf.indexed 1 thrpt 5 8582157,437 ┬▒ 215730,754 ops/s +/// StreamableFilterPerf.indexed 10 thrpt 5 3370403,716 ┬▒ 44180,727 ops/s +/// StreamableFilterPerf.indexed 100 thrpt 5 392725,429 ┬▒ 5825,883 ops/s +/// StreamableFilterPerf.indexed 1000 thrpt 5 42956,064 ┬▒ 774,701 ops/s +/// StreamableFilterPerf.indexed 10000 thrpt 5 4274,859 ┬▒ 48,781 ops/s +/// StreamableFilterPerf.indexed 100000 thrpt 5 427,545 ┬▒ 6,204 ops/s +/// StreamableFilterPerf.indexed 1000000 thrpt 5 38,761 ┬▒ 1,574 ops/s +/// ``` +/// +/// # 1. Avoid whenComplete and tidy up the internals, better wip management +/// +/// 1 times regression -22%, 1 million case +63% +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableFilterPerf.basic 1 thrpt 5 20450133,688 ┬▒ 292577,808 ops/s +/// StreamableFilterPerf.basic 10 thrpt 5 7001560,283 ┬▒ 95757,407 ops/s +/// StreamableFilterPerf.basic 100 thrpt 5 801954,196 ┬▒ 25583,620 ops/s +/// StreamableFilterPerf.basic 1000 thrpt 5 81090,359 ┬▒ 1422,785 ops/s +/// StreamableFilterPerf.basic 10000 thrpt 5 7861,791 ┬▒ 174,594 ops/s +/// StreamableFilterPerf.basic 100000 thrpt 5 791,253 ┬▒ 10,498 ops/s +/// StreamableFilterPerf.basic 1000000 thrpt 5 78,027 ┬▒ 1,068 ops/s +/// StreamableFilterPerf.enumerated 1 thrpt 5 11758684,924 ┬▒ 308081,461 ops/s +/// StreamableFilterPerf.enumerated 10 thrpt 5 4399154,841 ┬▒ 143732,147 ops/s +/// StreamableFilterPerf.enumerated 100 thrpt 5 581006,390 ┬▒ 15226,107 ops/s +/// StreamableFilterPerf.enumerated 1000 thrpt 5 52474,101 ┬▒ 900,684 ops/s +/// StreamableFilterPerf.enumerated 10000 thrpt 5 5364,237 ┬▒ 55,065 ops/s +/// StreamableFilterPerf.enumerated 100000 thrpt 5 575,620 ┬▒ 11,915 ops/s +/// StreamableFilterPerf.enumerated 1000000 thrpt 5 53,243 ┬▒ 0,953 ops/s +/// StreamableFilterPerf.indexed 1 thrpt 5 10378233,365 ┬▒ 320717,352 ops/s +/// StreamableFilterPerf.indexed 10 thrpt 5 4450787,764 ┬▒ 16570,500 ops/s +/// StreamableFilterPerf.indexed 100 thrpt 5 589007,872 ┬▒ 17229,304 ops/s +/// StreamableFilterPerf.indexed 1000 thrpt 5 61222,108 ┬▒ 1699,621 ops/s +/// StreamableFilterPerf.indexed 10000 thrpt 5 6181,191 ┬▒ 174,083 ops/s +/// StreamableFilterPerf.indexed 100000 thrpt 5 582,777 ┬▒ 14,846 ops/s +/// StreamableFilterPerf.indexed 1000000 thrpt 5 54,477 ┬▒ 1,366 ops/s +/// ``` +/// +/// # 2. Synchronous bias via CAS-based state management, via Claude Fable suggestions +/// +/// times-1 +70% vs baseline, times million +240% vs baseline +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableFilterPerf.basic 1 thrpt 5 44649205,329 ┬▒ 1318761,330 ops/s +/// StreamableFilterPerf.basic 10 thrpt 5 14573791,114 ┬▒ 307959,449 ops/s +/// StreamableFilterPerf.basic 100 thrpt 5 1704743,707 ┬▒ 27297,095 ops/s +/// StreamableFilterPerf.basic 1000 thrpt 5 169935,412 ┬▒ 1894,944 ops/s +/// StreamableFilterPerf.basic 10000 thrpt 5 16747,005 ┬▒ 149,331 ops/s +/// StreamableFilterPerf.basic 100000 thrpt 5 1630,870 ┬▒ 39,286 ops/s +/// StreamableFilterPerf.basic 1000000 thrpt 5 162,566 ┬▒ 4,036 ops/s +/// StreamableFilterPerf.enumerated 1 thrpt 5 13220834,886 ┬▒ 157990,717 ops/s +/// StreamableFilterPerf.enumerated 10 thrpt 5 6266989,179 ┬▒ 168298,570 ops/s +/// StreamableFilterPerf.enumerated 100 thrpt 5 894912,487 ┬▒ 4979,227 ops/s +/// StreamableFilterPerf.enumerated 1000 thrpt 5 89633,958 ┬▒ 1631,029 ops/s +/// StreamableFilterPerf.enumerated 10000 thrpt 5 8973,714 ┬▒ 197,897 ops/s +/// StreamableFilterPerf.enumerated 100000 thrpt 5 887,542 ┬▒ 12,736 ops/s +/// StreamableFilterPerf.enumerated 1000000 thrpt 5 88,537 ┬▒ 0,994 ops/s +/// StreamableFilterPerf.indexed 1 thrpt 5 11534659,877 ┬▒ 281862,191 ops/s +/// StreamableFilterPerf.indexed 10 thrpt 5 6663720,805 ┬▒ 127958,849 ops/s +/// StreamableFilterPerf.indexed 100 thrpt 5 973299,004 ┬▒ 35818,034 ops/s +/// StreamableFilterPerf.indexed 1000 thrpt 5 98248,858 ┬▒ 1445,809 ops/s +/// StreamableFilterPerf.indexed 10000 thrpt 5 9801,805 ┬▒ 183,524 ops/s +/// StreamableFilterPerf.indexed 100000 thrpt 5 983,497 ┬▒ 23,374 ops/s +/// StreamableFilterPerf.indexed 1000000 thrpt 5 97,605 ┬▒ 1,170 ops/s +/// ``` +/// +/// # 3. Add EnumerableSource and DeferredEnumerableSource paths. +/// +/// +193% times 1 vs baseline, +607% times million vs baseline +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableFilterPerf.basic 1 thrpt 5 44129567,500 ┬▒ 3907413,532 ops/s +/// StreamableFilterPerf.basic 10 thrpt 5 15360293,459 ┬▒ 134453,730 ops/s +/// StreamableFilterPerf.basic 100 thrpt 5 1811371,583 ┬▒ 32392,354 ops/s +/// StreamableFilterPerf.basic 1000 thrpt 5 163322,591 ┬▒ 4944,111 ops/s +/// StreamableFilterPerf.basic 10000 thrpt 5 16124,871 ┬▒ 578,538 ops/s +/// StreamableFilterPerf.basic 100000 thrpt 5 1618,494 ┬▒ 68,714 ops/s +/// StreamableFilterPerf.basic 1000000 thrpt 5 157,161 ┬▒ 1,401 ops/s +/// StreamableFilterPerf.enumerated 1 thrpt 5 27449227,121 ┬▒ 639347,534 ops/s +/// StreamableFilterPerf.enumerated 10 thrpt 5 43288619,923 ┬▒ 787987,036 ops/s +/// StreamableFilterPerf.enumerated 100 thrpt 5 3013969,186 ┬▒ 83719,137 ops/s +/// StreamableFilterPerf.enumerated 1000 thrpt 5 272763,930 ┬▒ 4591,739 ops/s +/// StreamableFilterPerf.enumerated 10000 thrpt 5 27699,074 ┬▒ 591,416 ops/s +/// StreamableFilterPerf.enumerated 100000 thrpt 5 2769,250 ┬▒ 75,561 ops/s +/// StreamableFilterPerf.enumerated 1000000 thrpt 5 264,589 ┬▒ 6,425 ops/s +/// StreamableFilterPerf.indexed 1 thrpt 5 11637720,830 ┬▒ 318248,700 ops/s +/// StreamableFilterPerf.indexed 10 thrpt 5 6651438,800 ┬▒ 179861,878 ops/s +/// StreamableFilterPerf.indexed 100 thrpt 5 978543,147 ┬▒ 34745,778 ops/s +/// StreamableFilterPerf.indexed 1000 thrpt 5 97398,438 ┬▒ 1071,570 ops/s +/// StreamableFilterPerf.indexed 10000 thrpt 5 9768,628 ┬▒ 221,269 ops/s +/// StreamableFilterPerf.indexed 100000 thrpt 5 972,717 ┬▒ 31,004 ops/s +/// StreamableFilterPerf.indexed 1000000 thrpt 5 95,972 ┬▒ 2,552 ops/s +/// ``` +@BenchmarkMode(Mode.Throughput) +@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@OutputTimeUnit(TimeUnit.SECONDS) +@Fork(value = 1) +@State(Scope.Thread) +public class StreamableFilterPerf { + @Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" }) + public int times; + + Streamable result; + Streamable> indexedMax; + Streamable> enumeratedMax; + + @Setup + public void setup() { + result = Streamable.range(1, times).filter(v -> (v & 1) != 0); + indexedMax = result.collect(Collectors.maxBy(Comparator.naturalOrder())); + enumeratedMax = Streamable.fromIterable(() -> IntStream.range(1, 1 + times).iterator()) + .filter(v -> (v & 1) != 0) + .collect(Collectors.maxBy(Comparator.naturalOrder())); + } + + @Benchmark + public Object basic() { + return result.blockingLast(); + } + + @Benchmark + public Object indexed() { + return indexedMax.blockingLast(); + } + + @Benchmark + public Object enumerated() { + return enumeratedMax.blockingLast(); + } +} diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java index 0fdee431b9..bdcba9b901 100644 --- a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java @@ -67,6 +67,21 @@ /// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 1408,117 ┬▒ 134,934 ops/s /// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 123,326 ┬▒ 29,358 ops/s /// ``` +/// +/// # 3. synchronous biased consumption loop, similar to filter, via Claude Fable +/// +/// +55% for times 1 vs baseline, +582% for times million vs baseline +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 22006265,959 ┬▒ 1776391,634 ops/s +/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 28047951,727 ┬▒ 302931,360 ops/s +/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 6137837,568 ┬▒ 100371,047 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 496011,671 ┬▒ 114738,714 ops/s +/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 51837,709 ┬▒ 3228,185 ops/s +/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 5094,499 ┬▒ 405,946 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 456,867 ┬▒ 15,351 ops/s +/// ``` @BenchmarkMode(Mode.Throughput) @Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) @Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java index c45204f3df..a1b3c9a7e0 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java @@ -13,7 +13,9 @@ package io.reactivex.rxjava4.internal.operators.streamable; +import java.io.Serial; import java.util.concurrent.*; +import java.util.concurrent.Future.State; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.*; import java.util.stream.Collector; @@ -41,9 +43,11 @@ public record StreamableCollector( } static final class CollectorStreamable + extends AtomicInteger implements Streamer, BiConsumer { - final AtomicInteger wip; + @Serial + private static final long serialVersionUID = -2893736592415047972L; final Streamer upstream; @@ -70,7 +74,6 @@ static final class CollectorStreamable Function finisher, StreamerCancellation cancellation) { this.upstream = upstream; - this.wip = new AtomicInteger(); this.storage = storage; this.accumulator = accumulator; this.finisher = finisher; @@ -170,32 +173,36 @@ void deferredEnumerate(boolean hasInitialValue, Throwable error) { } void drain() { - if (wip.getAndIncrement() != 0) { - return; - } - - int wipMax = 1; - int wipIndex = 0; - do { + for (;;) { if (done) { StreamableHelper.whenComplete(upstream.finish(), this); - break; + return; } else { - StreamableHelper.whenComplete(upstream.next(), this); - } - if (++wipIndex == wipMax) { - var newWip = wip.get(); - if (newWip != wipMax) { - wipMax = newWip; - } else { - wipMax = wip.addAndGet(-wipMax); - if (wipMax == 0) { - break; + var upstreamNext = upstream.next().toCompletableFuture(); + var state = upstreamNext.state(); + if (state == State.RUNNING) { + set(1); + upstreamNext.whenComplete(this); + if (compareAndSet(1, 0)) { + return; } - wipIndex = 0; + state = upstreamNext.state(); + } + + if (state == State.SUCCESS) { + if (upstreamNext.getNow((false))) { + accumulator.accept(storage, upstream.current()); + } else { + current = finisher.apply(storage); + nextReady.complete(true); + return; + } + } else { + nextReady.completeExceptionally(upstreamNext.exceptionNow()); + return; } } - } while (true); + } } @Override @@ -207,15 +214,17 @@ public void accept(Object t, Throwable u) { finishReady.complete(null); } } else { - if (u != null) { - nextReady.completeExceptionally(u); - } else - if ((Boolean)t) { - accumulator.accept(storage, upstream.current()); - drain(); - } else { - current = finisher.apply(storage); - nextReady.complete(true); + if (!compareAndSet(1, 2)) { + if (u != null) { + nextReady.completeExceptionally(u); + } else + if ((Boolean)t) { + accumulator.accept(storage, upstream.current()); + drain(); + } else { + current = finisher.apply(storage); + nextReady.complete(true); + } } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java index 98739de737..4083d862bd 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java @@ -14,11 +14,12 @@ package io.reactivex.rxjava4.internal.operators.streamable; import java.util.NoSuchElementException; -import java.util.concurrent.*; +import java.util.concurrent.CompletionStage; import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.operators.*; public enum StreamableEmpty implements Streamable { @@ -34,7 +35,7 @@ public static Streamer createEmpty() { return (Streamer)EmptyStreamer.INSTANCE; } - enum EmptyStreamer implements Streamer { + enum EmptyStreamer implements Streamer, IndexableSource, EnumerableSource { INSTANCE; @@ -52,5 +53,20 @@ enum EmptyStreamer implements Streamer { public @NonNull CompletionStage finish() { return FINISHED; } + + @Override + public boolean nextSync() throws Throwable { + return false; + } + + @Override + public @NonNull Object elementAt(long index) throws Throwable { + throw new NoSuchElementException("This Streamable/Streamer never has elements"); + } + + @Override + public long limit() { + return 0; + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilter.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilter.java index df405a1eb1..7df9dc58b6 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilter.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilter.java @@ -13,8 +13,11 @@ package io.reactivex.rxjava4.internal.operators.streamable; +import java.io.Serial; import java.util.concurrent.*; +import java.util.concurrent.Future.State; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; @@ -22,26 +25,45 @@ import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.Predicate; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.operators.*; public record StreamableFilter( @NonNull Streamable source, @NonNull Predicate predicate) implements Streamable, HasUpstreamStreamableSource { + @SuppressWarnings("unchecked") @Override public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { - return new FilterStreamer<>(source.stream(cancellation), predicate, cancellation); + var upstream = source.stream(cancellation); + // No IndexableSource because we don't know how many items would pass the predicate + // thus limit() would be non-calculatable + if (upstream instanceof DeferredEnumerableSource dsrc) { + return new FilterStreamerDeferredEnumerable<>(upstream, (DeferredEnumerableSource)dsrc, predicate, cancellation); + } else + if (upstream instanceof EnumerableSource esrc) { + return new FilterStreamerEnumerable<>(upstream, (EnumerableSource)esrc, predicate, cancellation); + } + return new FilterStreamerBasic<>(upstream, predicate, cancellation); } - static final class FilterStreamer implements Streamer { + static abstract class FilterStreamerBase extends AtomicInteger + implements Streamer, BiConsumer { + + @Serial + private static final long serialVersionUID = -4830414233351804049L; + final Streamer upstream; + final Predicate predicate; + StreamerCancellation cancellation; - volatile T current; - final AtomicInteger wip = new AtomicInteger(); + T current; + + CompletableFuture nextReady; - FilterStreamer(Streamer upstream, Predicate predicate, StreamerCancellation cancellation) { + FilterStreamerBase(Streamer upstream, Predicate predicate, StreamerCancellation cancellation) { this.upstream = upstream; this.cancellation = cancellation; this.predicate = predicate; @@ -49,9 +71,9 @@ static final class FilterStreamer implements Streamer { @Override public @NonNull CompletionStage next() { - var cf = new CompletableFuture(); - drain(cf); - return cf; + nextReady = new CompletableFuture(); + drain(); + return nextReady; } @Override @@ -66,35 +88,154 @@ static final class FilterStreamer implements Streamer { return upstream.finish(); } - void drain(CompletableFuture cf) { - if (wip.getAndIncrement() != 0) { - return; + void drain() { + for (;;) { + var upstreamNext = upstream.next().toCompletableFuture(); + + var state = upstreamNext.state(); + if (state == State.RUNNING) { + set(1); + upstreamNext.whenComplete(this); + if (compareAndSet(1, 0)) { + return; + } + state = upstreamNext.state(); + } + + if (state == State.SUCCESS) { + boolean has = upstreamNext.getNow(false); + if (!has) { + nextReady.complete(false); + return; + } + T value = upstream.current(); + boolean pass; + try { + pass = predicate.test(value); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + if (pass) { + current = value; + nextReady.complete(true); + return; + } + } else { + nextReady.completeExceptionally(upstreamNext.exceptionNow()); + return; + } } - do { - upstream.next() - .whenComplete((v, e) -> { - if (e != null) { - cf.completeExceptionally(e); - } else { - if (v) { - try { - var w = upstream.current(); - if (predicate.test(w)) { - current = w; - cf.complete(true); - } else { - drain(cf); - } - } catch (Throwable ex) { - Exceptions.throwIfFatal(ex); - cf.completeExceptionally(ex); - } + } + + @Override + public void accept(Boolean t, Throwable u) { + if (!compareAndSet(1, 2)) { + if (u != null) { + nextReady.completeExceptionally(u); + } else { + if (t) { + T value = upstream.current(); + boolean pass; + try { + pass = predicate.test(value); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + if (pass) { + current = value; + nextReady.complete(true); } else { - cf.complete(false); + drain(); } + } else { + nextReady.complete(false); } - }); - } while (wip.decrementAndGet() != 0); + } + } + } + } + + static final class FilterStreamerBasic extends FilterStreamerBase { + + @Serial + private static final long serialVersionUID = -7116891990338100320L; + + FilterStreamerBasic(Streamer upstream, + Predicate predicate, StreamerCancellation cancellation) { + super(upstream, predicate, cancellation); + } + } + + static final class FilterStreamerEnumerable extends FilterStreamerBase + implements EnumerableSource { + + @Serial + private static final long serialVersionUID = -7116891990338100320L; + + final EnumerableSource enumerable; + + FilterStreamerEnumerable(Streamer upstream, + EnumerableSource enumerable, + Predicate predicate, StreamerCancellation cancellation) { + super(upstream, predicate, cancellation); + this.enumerable = enumerable; + } + + @Override + public boolean nextSync() throws Throwable { + while (!cancellation.isDisposed()) { + if (enumerable.nextSync()) { + var value = enumerable.current(); + if (predicate.test(value)) { + current = value; + return true; + } + } else { + return false; + } + } + throw new CancellationException(); // FIXME maybe??? + } + } + + static final class FilterStreamerDeferredEnumerable extends FilterStreamerBase + implements DeferredEnumerableSource { + + @Serial + private static final long serialVersionUID = -7116891990338100320L; + + final DeferredEnumerableSource enumerable; + + FilterStreamerDeferredEnumerable(Streamer upstream, + DeferredEnumerableSource enumerable, + Predicate predicate, StreamerCancellation cancellation) { + super(upstream, predicate, cancellation); + this.enumerable = enumerable; + } + + @Override + public boolean nextSync() throws Throwable { + while (!cancellation.isDisposed()) { + if (enumerable.nextSync()) { + var value = enumerable.current(); + if (predicate.test(value)) { + current = value; + return true; + } + } else { + return false; + } + } + throw new CancellationException(); // FIXME maybe??? + } + + @Override + public CompletionStage enumerableReady() { + return enumerable.enumerableReady(); } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJust.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJust.java index e1ace46858..2bf50f9e2e 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJust.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJust.java @@ -18,6 +18,7 @@ import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.operators.*; public record StreamableJust(@NonNull T item) implements Streamable { @@ -26,11 +27,12 @@ public record StreamableJust(@NonNull T item) implements Streamable { return new JustStreamer<>(item); } - static final class JustStreamer implements Streamer { + static final class JustStreamer + implements Streamer, IndexableSource, EnumerableSource { - volatile T item; + final T item; - volatile int stage; + int stage; JustStreamer(T item) { this.item = item; @@ -42,8 +44,6 @@ static final class JustStreamer implements Streamer { stage = 1; return NEXT_TRUE; } - item = null; - stage = 2; return NEXT_FALSE; } @@ -54,9 +54,22 @@ static final class JustStreamer implements Streamer { @Override public @NonNull CompletionStage finish() { - item = null; - stage = 2; return FINISHED; } + + @Override + public @NonNull T elementAt(long index) throws Throwable { + return item; + } + + @Override + public long limit() { + return 1; + } + + @Override + public boolean nextSync() throws Throwable { + return stage++ == 0; + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java index b42620b635..7c18122f27 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java @@ -15,6 +15,7 @@ import java.io.Serial; import java.util.NoSuchElementException; +import java.util.concurrent.Future.State; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; @@ -82,29 +83,40 @@ static final class LastStreamer } void drain() { - if (getAndIncrement() != 0) { - return; - } - - int wipMax = 1; - int wipIndex = 0; - do { + for (;;) { if (done) { StreamableHelper.whenComplete(upstream.finish(), this); - break; + return; } else { - StreamableHelper.whenComplete(upstream.next(), this); - } - if (++wipIndex == wipMax) { - wipMax = get(); - if (wipIndex == wipMax) { - wipMax = addAndGet(-wipMax); - if (wipMax != 0) { - wipIndex = 0; + var upstreamNext = upstream.next().toCompletableFuture(); + + var state = upstreamNext.state(); + + if (state == State.RUNNING) { + set(1); + upstreamNext.whenComplete(this); + if (compareAndSet(1, 0)) { + return; } + state = upstreamNext.state(); + } + + if (state == State.SUCCESS) { + if (upstreamNext.getNow(false)) { + current = upstream.current(); + } else { + done = true; + StreamableHelper.whenComplete(upstream.finish(), this); + return; + } + } else { + nextFailure = upstreamNext.exceptionNow(); + done = true; + StreamableHelper.whenComplete(upstream.finish(), this); + return; } } - } while (wipMax != 0); + } } @Override @@ -124,17 +136,19 @@ public void accept(Object t, Throwable u) { } } } else { - if (u != null) { - nextFailure = u; - done = true; - drain(); - } else { - if ((Boolean)t) { - current = upstream.current(); - } else { + if (!compareAndSet(1, 2)) { + if (u != null) { + nextFailure = u; done = true; + drain(); + } else { + if ((Boolean)t) { + current = upstream.current(); + } else { + done = true; + } + drain(); } - drain(); } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMap.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMap.java index aecd9f14f9..3e59cdd50a 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMap.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMap.java @@ -37,10 +37,10 @@ public record StreamableMap( var upstream = source.stream(cancellation); if (upstream instanceof IndexableSource isrc) { return new MapStreamerIndexed<>(upstream, (IndexableSource)isrc, mapper); - } + } else if (upstream instanceof DeferredEnumerableSource esrc) { return new MapStreamerDeferredEnumerated<>(upstream, (DeferredEnumerableSource)esrc, mapper); - } + } else if (upstream instanceof EnumerableSource esrc) { return new MapStreamerEnumerated<>(upstream, (EnumerableSource)esrc, mapper); } @@ -114,10 +114,12 @@ static final class MapStreamerBasic extends MapStreamerBase { } static final class MapStreamerIndexed extends MapStreamerBase - implements IndexableSource { + implements IndexableSource, EnumerableSource { final IndexableSource indexed; + long index = -1L; + MapStreamerIndexed(Streamer upstream, IndexableSource indexed, Function mapper) { super(upstream, mapper); this.indexed = indexed; @@ -132,6 +134,15 @@ static final class MapStreamerIndexed extends MapStreamerBase public long limit() { return indexed.limit(); } + + @Override + public boolean nextSync() throws Throwable { + if (++index >= indexed.limit()) { + return false; + } + current = elementAt(index); + return true; + } } static final class MapStreamerEnumerated extends MapStreamerBase diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRange.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRange.java index 08d9bd025f..43f0e6783a 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRange.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRange.java @@ -18,7 +18,7 @@ import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.StreamerCancellation; -import io.reactivex.rxjava4.operators.IndexableSource; +import io.reactivex.rxjava4.operators.*; public record StreamableRange(int start, int count) implements Streamable { @@ -27,7 +27,8 @@ public record StreamableRange(int start, int count) implements Streamable(start, start + count); } - static final class RangeStreamer implements Streamer, IndexableSource { + static final class RangeStreamer + implements Streamer, IndexableSource, EnumerableSource { final int start; @@ -69,5 +70,13 @@ public Integer elementAt(long index) { public long limit() { return end - start; } + + @Override + public boolean nextSync() throws Throwable { + if (++current >= end) { + return false; + } + return true; + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeLong.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeLong.java index 397ac4f3f4..eec67d1921 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeLong.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeLong.java @@ -18,6 +18,7 @@ import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.operators.*; public record StreamableRangeLong(long start, long count) implements Streamable { @@ -26,7 +27,10 @@ public record StreamableRangeLong(long start, long count) implements Streamable< return new RangeLongStreamer<>(start, start + count); } - static final class RangeLongStreamer implements Streamer { + static final class RangeLongStreamer + implements Streamer, IndexableSource, EnumerableSource { + + final long start; final long end; @@ -36,6 +40,7 @@ static final class RangeLongStreamer implements Streamer { RangeLongStreamer(long start, long end) { index = start; + this.start = start; this.end = end; } @@ -61,5 +66,23 @@ static final class RangeLongStreamer implements Streamer { current = end; return FINISHED; } + + @Override + public Long elementAt(long index) { + return start + index; + } + + @Override + public long limit() { + return end - start; + } + + @Override + public boolean nextSync() throws Throwable { + if (++current >= end) { + return false; + } + return true; + } } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java index 80003015ce..1d811efcca 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java @@ -24,6 +24,8 @@ import io.reactivex.rxjava4.core.Streamable; import io.reactivex.rxjava4.disposables.CompositeDisposable; import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableCollectorTest extends StreamableBaseTest { @@ -85,4 +87,56 @@ public void indexerDisposed() throws Throwable { .awaitNext(); }); } + + @Test + public void intelvalRange() throws Throwable { + Streamable.intervalRange(1, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()) + .hide() + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1L, 2L, 3L, 4L, 5L)); + } + + @Test + public void viaDispatch() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp + .collect(Collectors.toList()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + dsp.next(2).toCompletableFuture().join(); + dsp.next(3).toCompletableFuture().join(); + dsp.finish(null).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3)); + } + + @Test + public void viaDispatchError() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp + .collect(Collectors.toList()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + dsp.next(2).toCompletableFuture().join(); + dsp.next(3).toCompletableFuture().join(); + dsp.finish(new TestException()).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java index a543476a47..78aab29b9e 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java @@ -304,4 +304,20 @@ public void virtualCreateFinishFail() { .awaitDone(5, TimeUnit.SECONDS) .assertFailure(TestException.class, 1, 2, 3, 4, 5); } + + @Test + public void delay() throws Throwable { + Streamable.concat(List.of( + Streamable.range(1, 5), + Streamable.range(6, 5).delay(1, TimeUnit.MILLISECONDS, Schedulers.single()), + Streamable.range(11, 5))) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult( + 1, 2, 3, 4, 5, + 6, 7, 8, 9, 10, + 11, 12, 13, 14, 15 + ); + } + } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java new file mode 100644 index 0000000000..eb675152fc --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java @@ -0,0 +1,67 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.*; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; + +public class StreamableEmptyTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.empty() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void collectIndexer() throws Throwable { + Streamable.empty() + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of()); + } + + @Test + public void collectEnumerator() throws Throwable { + Streamable.empty() + .filter(_ -> true) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of()); + } + + @Test + public void noSuchElementExceptionCurrent() { + assertThrows(NoSuchElementException.class, () -> { + StreamableEmpty.EmptyStreamer.INSTANCE.current(); + }); + } + @Test + public void noSuchElementExceptionElementAt() { + assertThrows(NoSuchElementException.class, () -> { + StreamableEmpty.EmptyStreamer.INSTANCE.elementAt(0L); + }); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilterTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilterTest.java index 03f1aa0c4a..dbb8797cfa 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilterTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilterTest.java @@ -13,11 +13,15 @@ package io.reactivex.rxjava4.internal.operators.streamable; +import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; -import io.reactivex.rxjava4.core.Streamable; + +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableFilterTest extends StreamableBaseTest { @@ -95,4 +99,193 @@ public void sourceError2() { .assertFailure(TestException.class); } + @Test + public void delayed() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void delayed2() { + Streamable.range(1, 100) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .filter(_ -> false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void longSync() { + Streamable.range(1, 1_000_000) + .filter(v -> (v & 1) != 0) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertValueCount(500_000) + .assertNoErrors() + .assertComplete() + ; + } + + @Test + public void delayedCrash() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .filter(_ -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void delayed3() { + Streamable.range(1, 100) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .filter(_ -> true) + .filter(_ -> false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void timeout1() { + Streamable.never() + .timeout(1, TimeUnit.MILLISECONDS, Schedulers.single(), Streamable.empty()) + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void timeout2() { + Streamable.never() + .timeout(1, TimeUnit.MILLISECONDS, Schedulers.single(), Streamable.error(new TestException())) + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void basicHidden() { + Streamable.range(1, 5) + .hide() + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void basicHidden2() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .hide() + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void indexable() { + Streamable.range(1, 5) + .filter(_ -> true) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void indexable2() { + Streamable.range(1, 5) + .filter(_ -> true) + .collect(Collectors.toList()) + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void enumerable3() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .filter(_ -> true) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void enumerable4() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .filter(_ -> true) + .collect(Collectors.toList()) + .filter(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void enumerable5() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .filter(_ -> false) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of()); + } + + @Test + public void enumerable6() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .filter(_ -> false) + .collect(Collectors.toList()) + .filter(_ -> false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void enumerable7() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .filter(_ -> false) + .collect(Collectors.toList()) + .filter(_ -> false) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of()); + } + + @Test + public void deferredEnumerable() { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, v + 1, v + 2, v + 3, v + 4)) + .filter(_ -> true) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void deferredEnumerable2() { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, v + 1, v + 2, v + 3, v + 4)) + .filter(v -> v % 2 == 0) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(2, 4)); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJustTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJustTest.java new file mode 100644 index 0000000000..b3086983b0 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJustTest.java @@ -0,0 +1,53 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; + +public class StreamableJustTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.just(1) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + + @Test + public void collectIndexer() throws Throwable { + Streamable.just(1) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1)); + } + + @Test + public void collectEnumerator() throws Throwable { + Streamable.just(1) + .filter(_ -> true) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1)); + } + +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java index f62bf4911f..0fc824f8a2 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java @@ -24,6 +24,7 @@ import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; import io.reactivex.rxjava4.processors.DispatchStreamProcessor; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableLastAsSingleTest extends StreamableBaseTest { @@ -106,4 +107,13 @@ public void hasSource() { assertTrue(operator instanceof HasUpstreamStreamableSource huss && source == huss.source(), "HasUpstreamStreamableSource not supported or source() returns something unexpected: " + operator); } + + @Test + public void intervalRange() throws Throwable { + Streamable.intervalRange(1, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()) + .lastOrError() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(5L); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapTest.java index 11bf6d4414..65fd0622c5 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapTest.java @@ -213,4 +213,14 @@ public void deferredEnumerableDebug() throws Throwable { }); } + @Test + public void indexedToEnumerable() throws Throwable { + Streamable.range(1, 5) + .map(v -> v + 1) + .filter(v -> v % 2 == 0) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(2, 4, 6)); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeTest.java index 4efa7c35a0..0b55626c17 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeTest.java @@ -15,9 +15,12 @@ import static org.junit.jupiter.api.Assertions.assertThrows; +import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; + import io.reactivex.rxjava4.core.Streamable; public class StreamableRangeTest extends StreamableBaseTest { @@ -102,4 +105,23 @@ public void underOverflowLong() throws Throwable { public void underNoOverflowLong() throws Throwable { Streamable.rangeLong(-2, Long.MAX_VALUE); } + + @Test + public void longIndexed() throws Throwable { + Streamable.rangeLong(1, 3) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1L, 2L, 3L)); + } + + @Test + public void longEnumerated() throws Throwable { + Streamable.rangeLong(1, 3) + .filter(_ -> true) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1L, 2L, 3L)); + } } From 0712af621fad85760733b92036f0909c72f7a649 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Sat, 18 Jul 2026 11:52:14 +0200 Subject: [PATCH 2/2] Style fix --- .../internal/operators/streamable/StreamableEmptyTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java index eb675152fc..b3a2fa2698 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmptyTest.java @@ -58,6 +58,7 @@ public void noSuchElementExceptionCurrent() { StreamableEmpty.EmptyStreamer.INSTANCE.current(); }); } + @Test public void noSuchElementExceptionElementAt() { assertThrows(NoSuchElementException.class, () -> {