diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableConcatIterablePerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableConcatIterablePerf.java new file mode 100644 index 0000000000..307556a486 --- /dev/null +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableConcatIterablePerf.java @@ -0,0 +1,149 @@ +/* + * 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.List; +import java.util.concurrent.TimeUnit; + +import org.openjdk.jmh.annotations.*; + +import io.reactivex.rxjava4.core.Streamable; + +/// +/// The concat(Iterable) seems to be one of the high allocators in the Scrabble benchmark +/// because of the continuation probably? +/// +/// i9 275HX, 32GB LPDDR5 6400MT CL52, Windows 25H2, JDK 26.0.1 +/// +/// # 0. Baseline +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 3474744,064 ┬▒ 92593,235 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1640843,357 ┬▒ 30500,201 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 236067,804 ┬▒ 1324,984 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 24348,807 ┬▒ 494,837 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 2469,689 ┬▒ 30,614 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 241,203 ┬▒ 5,330 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 24,330 ┬▒ 0,886 ops/s +/// ``` +/// +/// # 1. Reduce allocation in whenComplete +/// +/// No practical effect +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 3517630,344 ┬▒ 203989,280 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1633567,521 ┬▒ 40147,286 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 236815,630 ┬▒ 6185,836 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 23939,185 ┬▒ 319,632 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 2452,654 ┬▒ 98,012 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 239,677 ┬▒ 6,460 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 24,930 ┬▒ 1,884 ops/s +/// ``` +/// +/// # 2. avoid calling `whenComplete` +/// +/// +35% performance vs 1 for longer sequences +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 3785671,506 ┬▒ 124882,806 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 2028842,852 ┬▒ 76338,957 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 300832,578 ┬▒ 9916,135 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 33103,046 ┬▒ 856,542 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3272,894 ┬▒ 129,658 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 335,497 ┬▒ 4,678 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 33,685 ┬▒ 0,565 ops/s +/// ``` +/// +/// # 3. avoid calling decrementAndGet every time on a synchronous/reentrant usage +/// +/// +6% for the 1 case but, -5% performance regression vs optimization 2 +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 4021165,636 ┬▒ 93456,776 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1990327,224 ┬▒ 24421,179 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 301872,212 ┬▒ 2880,776 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 30909,533 ┬▒ 620,143 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3174,424 ┬▒ 37,618 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 316,643 ┬▒ 4,920 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 31,970 ┬▒ 0,521 ops/s +/// ``` +/// +/// # 4. restore the decrementAndGet use +/// +/// +/- 1% vs optimization 3 +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 4034863,551 ┬▒ 147716,809 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1951987,827 ┬▒ 29883,362 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 302013,530 ┬▒ 4251,994 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 33260,723 ┬▒ 755,076 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3329,779 ┬▒ 81,860 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 333,358 ┬▒ 8,474 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 32,943 ┬▒ 1,187 ops/s +/// ``` +/// # 5. fall through to getting the first item from the next source when it is picked, save a drain call +/// +/// +5% for short sequences, +22% for short sequences +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 4252027,061 ┬▒ 65983,230 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 2125263,936 ┬▒ 46999,223 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 304414,571 ┬▒ 3778,558 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 32123,342 ┬▒ 915,151 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3328,728 ┬▒ 38,444 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 331,795 ┬▒ 4,418 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 33,268 ┬▒ 0,899 ops/s +/// ``` +/// +/// # 6. synchronous-biased next() +/// +/// +77% short sequences, +762% for long sequences +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 6172450,906 ┬▒ 66554,843 ops/s +/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 5367812,331 ┬▒ 115663,523 ops/s +/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 1820204,546 ┬▒ 143049,591 ops/s +/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 218017,883 ┬▒ 49050,865 ops/s +/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 17406,789 ┬▒ 1079,512 ops/s +/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 2103,477 ┬▒ 360,090 ops/s +/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 207,888 ┬▒ 22,014 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 StreamableConcatIterablePerf { + @Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" }) + public int times; + + Streamable result; + + @Setup + public void setup() { + result = Streamable.concat(List.of(Streamable.range(1, times), Streamable.range(times + 1, times))); + } + + @Benchmark + public Object benchmark() { + return result.blockingLast(); + } +} diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java new file mode 100644 index 0000000000..0fdee431b9 --- /dev/null +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java @@ -0,0 +1,91 @@ +/* + * 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.concurrent.TimeUnit; + +import org.openjdk.jmh.annotations.*; + +import io.reactivex.rxjava4.core.*; + +/// +/// 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 +/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 14165950,142 ┬▒ 277135,092 ops/s +/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 5437536,431 ┬▒ 89291,499 ops/s +/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 688500,242 ┬▒ 9755,649 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 67446,091 ┬▒ 4208,462 ops/s +/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 6880,385 ┬▒ 223,058 ops/s +/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 690,980 ┬▒ 29,543 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 66,935 ┬▒ 1,086 ops/s +/// ``` +/// +/// # 1. avoid whenComplete +/// +/// +17% for times 1, +50% for a million +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 16684817,192 ┬▒ 337671,603 ops/s +/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 7713698,470 ┬▒ 72192,494 ops/s +/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 992063,105 ┬▒ 31518,477 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 105383,988 ┬▒ 4062,065 ops/s +/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 10103,558 ┬▒ 494,420 ops/s +/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 1022,181 ┬▒ 38,701 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 100,744 ┬▒ 1,390 ops/s +/// ``` +/// +/// # 2. batch wip accounting +/// +/// +6.8% for times 1 vs optimization 1. +23.3% for times million vs #0 +/// +25.8% for times 1 vs baseline, +82% for times million vs baseline +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 17823989,674 ┬▒ 243372,183 ops/s +/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 9319936,731 ┬▒ 157326,848 ops/s +/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 1308207,680 ┬▒ 14246,778 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 138722,075 ┬▒ 29515,013 ops/s +/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 11938,692 ┬▒ 354,381 ops/s +/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 1408,117 ┬▒ 134,934 ops/s +/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 123,326 ┬▒ 29,358 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 StreamableLastAsSinglePerf { + @Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" }) + public int times; + + Single result; + + @Setup + public void setup() { + result = Streamable.range(1, times).lastOrError(); + } + + @Benchmark + public Object benchmark() { + return result.blockingGet(); + } +} diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapPerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapPerf.java new file mode 100644 index 0000000000..131b4404b3 --- /dev/null +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapPerf.java @@ -0,0 +1,209 @@ +/* + * 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 +/// StreamableMapPerf.basic 1 thrpt 5 596170000,249 ┬▒ 58386686,173 ops/s +/// StreamableMapPerf.basic 10 thrpt 5 88996451,855 ┬▒ 7879743,705 ops/s +/// StreamableMapPerf.basic 100 thrpt 5 12136698,201 ┬▒ 1117754,568 ops/s +/// StreamableMapPerf.basic 1000 thrpt 5 183021,386 ┬▒ 7623,362 ops/s +/// StreamableMapPerf.basic 10000 thrpt 5 17876,769 ┬▒ 1359,162 ops/s +/// StreamableMapPerf.basic 100000 thrpt 5 1809,422 ┬▒ 31,426 ops/s +/// StreamableMapPerf.basic 1000000 thrpt 5 175,036 ┬▒ 11,687 ops/s +/// StreamableMapPerf.enumerated 1 thrpt 5 16123906,915 ┬▒ 1069401,973 ops/s +/// StreamableMapPerf.enumerated 10 thrpt 5 5693840,612 ┬▒ 66583,096 ops/s +/// StreamableMapPerf.enumerated 100 thrpt 5 761972,822 ┬▒ 23829,829 ops/s +/// StreamableMapPerf.enumerated 1000 thrpt 5 74078,580 ┬▒ 9930,126 ops/s +/// StreamableMapPerf.enumerated 10000 thrpt 5 7340,915 ┬▒ 1049,583 ops/s +/// StreamableMapPerf.enumerated 100000 thrpt 5 679,864 ┬▒ 11,764 ops/s +/// StreamableMapPerf.enumerated 1000000 thrpt 5 70,816 ┬▒ 0,515 ops/s +/// StreamableMapPerf.indexed 1 thrpt 5 11560646,034 ┬▒ 1034530,694 ops/s +/// StreamableMapPerf.indexed 10 thrpt 5 5607940,249 ┬▒ 305032,627 ops/s +/// StreamableMapPerf.indexed 100 thrpt 5 804783,705 ┬▒ 12452,982 ops/s +/// StreamableMapPerf.indexed 1000 thrpt 5 79891,842 ┬▒ 8009,334 ops/s +/// StreamableMapPerf.indexed 10000 thrpt 5 8005,019 ┬▒ 894,673 ops/s +/// StreamableMapPerf.indexed 100000 thrpt 5 800,412 ┬▒ 94,950 ops/s +/// StreamableMapPerf.indexed 1000000 thrpt 5 80,535 ┬▒ 1,576 ops/s +/// ``` +/// +/// # 1. remove current volatile, avoid whenComplete calls when possible +/// +/// +86% on the longest basic, +45% on the indexed case (but no fusion yet), +/// -24% on the single element case. +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableMapPerf.basic 1 thrpt 5 455082027,159 ┬▒ 12415682,459 ops/s +/// StreamableMapPerf.basic 10 thrpt 5 81956627,142 ┬▒ 12197594,471 ops/s +/// StreamableMapPerf.basic 100 thrpt 5 11348376,707 ┬▒ 812958,859 ops/s +/// StreamableMapPerf.basic 1000 thrpt 5 319588,236 ┬▒ 56712,576 ops/s +/// StreamableMapPerf.basic 10000 thrpt 5 33363,038 ┬▒ 12143,283 ops/s +/// StreamableMapPerf.basic 100000 thrpt 5 3237,479 ┬▒ 308,847 ops/s +/// StreamableMapPerf.basic 1000000 thrpt 5 326,468 ┬▒ 57,581 ops/s +/// StreamableMapPerf.enumerated 1 thrpt 5 16730548,276 ┬▒ 1509643,233 ops/s +/// StreamableMapPerf.enumerated 10 thrpt 5 7214156,686 ┬▒ 204146,871 ops/s +/// StreamableMapPerf.enumerated 100 thrpt 5 1020491,947 ┬▒ 108128,423 ops/s +/// StreamableMapPerf.enumerated 1000 thrpt 5 99493,896 ┬▒ 3355,031 ops/s +/// StreamableMapPerf.enumerated 10000 thrpt 5 10197,248 ┬▒ 809,997 ops/s +/// StreamableMapPerf.enumerated 100000 thrpt 5 1050,388 ┬▒ 85,799 ops/s +/// StreamableMapPerf.enumerated 1000000 thrpt 5 99,972 ┬▒ 0,566 ops/s +/// StreamableMapPerf.indexed 1 thrpt 5 14901677,115 ┬▒ 234094,126 ops/s +/// StreamableMapPerf.indexed 10 thrpt 5 7654935,342 ┬▒ 54572,642 ops/s +/// StreamableMapPerf.indexed 100 thrpt 5 1198988,599 ┬▒ 211554,095 ops/s +/// StreamableMapPerf.indexed 1000 thrpt 5 115874,465 ┬▒ 12076,343 ops/s +/// StreamableMapPerf.indexed 10000 thrpt 5 13358,450 ┬▒ 286,266 ops/s +/// StreamableMapPerf.indexed 100000 thrpt 5 1221,610 ┬▒ 268,159 ops/s +/// StreamableMapPerf.indexed 1000000 thrpt 5 116,228 ┬▒ 1,277 ops/s +/// ``` +/// +/// # 2. fuse with indexable source +/// +/// +158% improvement on the longest chain. +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableMapPerf.basic 1 thrpt 5 458349718,751 ┬▒ 34799991,230 ops/s +/// StreamableMapPerf.basic 10 thrpt 5 77759584,162 ┬▒ 484470,610 ops/s +/// StreamableMapPerf.basic 100 thrpt 5 10927029,754 ┬▒ 439513,962 ops/s +/// StreamableMapPerf.basic 1000 thrpt 5 486654,649 ┬▒ 28948,227 ops/s +/// StreamableMapPerf.basic 10000 thrpt 5 39169,073 ┬▒ 1954,250 ops/s +/// StreamableMapPerf.basic 100000 thrpt 5 3823,117 ┬▒ 128,942 ops/s +/// StreamableMapPerf.basic 1000000 thrpt 5 385,872 ┬▒ 83,309 ops/s +/// StreamableMapPerf.enumerated 1 thrpt 5 17568827,861 ┬▒ 1188973,311 ops/s +/// StreamableMapPerf.enumerated 10 thrpt 5 7346307,943 ┬▒ 62971,019 ops/s +/// StreamableMapPerf.enumerated 100 thrpt 5 974720,167 ┬▒ 55352,752 ops/s +/// StreamableMapPerf.enumerated 1000 thrpt 5 101154,336 ┬▒ 10030,194 ops/s +/// StreamableMapPerf.enumerated 10000 thrpt 5 10020,188 ┬▒ 154,534 ops/s +/// StreamableMapPerf.enumerated 100000 thrpt 5 1005,707 ┬▒ 15,859 ops/s +/// StreamableMapPerf.enumerated 1000000 thrpt 5 98,836 ┬▒ 5,676 ops/s +/// StreamableMapPerf.indexed 1 thrpt 5 14395237,213 ┬▒ 648713,966 ops/s +/// StreamableMapPerf.indexed 10 thrpt 5 60821106,582 ┬▒ 1870263,863 ops/s +/// StreamableMapPerf.indexed 100 thrpt 5 3844820,803 ┬▒ 714985,119 ops/s +/// StreamableMapPerf.indexed 1000 thrpt 5 288269,160 ┬▒ 33882,613 ops/s +/// StreamableMapPerf.indexed 10000 thrpt 5 29448,704 ┬▒ 3040,234 ops/s +/// StreamableMapPerf.indexed 100000 thrpt 5 2941,029 ┬▒ 77,884 ops/s +/// StreamableMapPerf.indexed 1000000 thrpt 5 300,212 ┬▒ 7,796 ops/s +/// ``` +/// +/// # 3. use CompletableFuture.state to instead of isDone + isCompletedExceptionally overhead +/// +/// +7.7% for times 1, +12.5% for times 10. +/// Still -18% relative to the baseline. +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableMapPerf.basic 1 thrpt 5 493964135,333 ┬▒ 31919189,421 ops/s +/// StreamableMapPerf.basic 10 thrpt 5 87476563,366 ┬▒ 37519691,468 ops/s +/// StreamableMapPerf.basic 100 thrpt 5 10107439,791 ┬▒ 267335,311 ops/s +/// StreamableMapPerf.basic 1000 thrpt 5 523564,103 ┬▒ 105090,078 ops/s +/// StreamableMapPerf.basic 10000 thrpt 5 36818,972 ┬▒ 1944,375 ops/s +/// StreamableMapPerf.basic 100000 thrpt 5 3859,654 ┬▒ 46,790 ops/s +/// StreamableMapPerf.basic 1000000 thrpt 5 379,326 ┬▒ 43,010 ops/s +/// StreamableMapPerf.enumerated 1 thrpt 5 17760226,834 ┬▒ 1599737,873 ops/s +/// StreamableMapPerf.enumerated 10 thrpt 5 7297322,361 ┬▒ 105078,437 ops/s +/// StreamableMapPerf.enumerated 100 thrpt 5 1006007,580 ┬▒ 121150,559 ops/s +/// StreamableMapPerf.enumerated 1000 thrpt 5 101828,572 ┬▒ 9235,915 ops/s +/// StreamableMapPerf.enumerated 10000 thrpt 5 10712,997 ┬▒ 271,737 ops/s +/// StreamableMapPerf.enumerated 100000 thrpt 5 1001,925 ┬▒ 39,411 ops/s +/// StreamableMapPerf.enumerated 1000000 thrpt 5 100,114 ┬▒ 3,677 ops/s +/// StreamableMapPerf.indexed 1 thrpt 5 14962166,447 ┬▒ 1351752,478 ops/s +/// StreamableMapPerf.indexed 10 thrpt 5 18176023,720 ┬▒ 799380,855 ops/s +/// StreamableMapPerf.indexed 100 thrpt 5 2947669,689 ┬▒ 108127,841 ops/s +/// StreamableMapPerf.indexed 1000 thrpt 5 282842,159 ┬▒ 47754,366 ops/s +/// StreamableMapPerf.indexed 10000 thrpt 5 28008,555 ┬▒ 1513,803 ops/s +/// StreamableMapPerf.indexed 100000 thrpt 5 2830,012 ┬▒ 67,427 ops/s +/// StreamableMapPerf.indexed 1000000 thrpt 5 290,707 ┬▒ 25,138 ops/s +/// ``` +/// +/// # 4. enumerable fusion +/// +/// +37% on small times, +60% throughput on large times +/// ``` +/// enchmark (times) Mode Cnt Score Error Units +/// StreamableMapPerf.basic 1 thrpt 5 471719643,711 ┬▒ 106189283,745 ops/s +/// StreamableMapPerf.basic 10 thrpt 5 85398301,091 ┬▒ 17409080,046 ops/s +/// StreamableMapPerf.basic 100 thrpt 5 10869015,976 ┬▒ 1011070,084 ops/s +/// StreamableMapPerf.basic 1000 thrpt 5 549938,103 ┬▒ 42822,156 ops/s +/// StreamableMapPerf.basic 10000 thrpt 5 42652,848 ┬▒ 4870,733 ops/s +/// StreamableMapPerf.basic 100000 thrpt 5 4219,335 ┬▒ 905,798 ops/s +/// StreamableMapPerf.basic 1000000 thrpt 5 420,184 ┬▒ 77,257 ops/s +/// StreamableMapPerf.enumerated 1 thrpt 5 24390785,396 ┬▒ 748866,046 ops/s +/// StreamableMapPerf.enumerated 10 thrpt 5 32365811,876 ┬▒ 2323645,881 ops/s +/// StreamableMapPerf.enumerated 100 thrpt 5 2091169,420 ┬▒ 255405,462 ops/s +/// StreamableMapPerf.enumerated 1000 thrpt 5 182290,531 ┬▒ 9181,593 ops/s +/// StreamableMapPerf.enumerated 10000 thrpt 5 17774,650 ┬▒ 474,070 ops/s +/// StreamableMapPerf.enumerated 100000 thrpt 5 1774,772 ┬▒ 112,585 ops/s +/// StreamableMapPerf.enumerated 1000000 thrpt 5 164,270 ┬▒ 5,778 ops/s +/// StreamableMapPerf.indexed 1 thrpt 5 14900664,969 ┬▒ 213717,815 ops/s +/// StreamableMapPerf.indexed 10 thrpt 5 63037359,537 ┬▒ 1085133,528 ops/s +/// StreamableMapPerf.indexed 100 thrpt 5 3160805,643 ┬▒ 883774,226 ops/s +/// StreamableMapPerf.indexed 1000 thrpt 5 284112,887 ┬▒ 50755,549 ops/s +/// StreamableMapPerf.indexed 10000 thrpt 5 27770,464 ┬▒ 1669,842 ops/s +/// StreamableMapPerf.indexed 100000 thrpt 5 2966,157 ┬▒ 181,954 ops/s +/// StreamableMapPerf.indexed 1000000 thrpt 5 286,102 ┬▒ 52,939 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 StreamableMapPerf { + @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).map(v -> v + 1); + indexedMax = result.collect(Collectors.maxBy(Comparator.naturalOrder())); + enumeratedMax = Streamable.fromIterable(() -> IntStream.range(1, 1 + times).iterator()) + .map(v -> v + 1) + .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/StreamableSkipPerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java new file mode 100644 index 0000000000..1f617c559d --- /dev/null +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java @@ -0,0 +1,90 @@ +/* + * 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.concurrent.TimeUnit; + +import org.openjdk.jmh.annotations.*; + +import io.reactivex.rxjava4.core.*; + +/// +/// 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 +/// StreamableSkipPerf.benchmark 1 thrpt 5 1414799252,380 ┬▒ 237083836,864 ops/s +/// StreamableSkipPerf.benchmark 10 thrpt 5 10427857,800 ┬▒ 458808,883 ops/s +/// StreamableSkipPerf.benchmark 100 thrpt 5 1242428,922 ┬▒ 11522,536 ops/s +/// StreamableSkipPerf.benchmark 1000 thrpt 5 123656,684 ┬▒ 1351,735 ops/s +/// StreamableSkipPerf.benchmark 10000 thrpt 5 12247,572 ┬▒ 250,378 ops/s +/// StreamableSkipPerf.benchmark 100000 thrpt 5 1213,043 ┬▒ 21,316 ops/s +/// StreamableSkipPerf.benchmark 1000000 thrpt 5 124,518 ┬▒ 2,140 ops/s +/// ``` +/// +/// # 1. avoid whenComplete +/// +/// +39% for times 10 and million +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableSkipPerf.benchmark 1 thrpt 5 1354250907,646 ┬▒ 261597823,396 ops/s +/// StreamableSkipPerf.benchmark 10 thrpt 5 14537792,082 ┬▒ 941666,637 ops/s +/// StreamableSkipPerf.benchmark 100 thrpt 5 1921936,485 ┬▒ 35720,007 ops/s +/// StreamableSkipPerf.benchmark 1000 thrpt 5 164997,708 ┬▒ 2391,852 ops/s +/// StreamableSkipPerf.benchmark 10000 thrpt 5 17398,539 ┬▒ 660,608 ops/s +/// StreamableSkipPerf.benchmark 100000 thrpt 5 1719,680 ┬▒ 22,035 ops/s +/// StreamableSkipPerf.benchmark 1000000 thrpt 5 173,791 ┬▒ 4,496 ops/s +/// ``` +/// +/// # 2. batch wip accounting +/// +/// +74% for times 10, +72% for million +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableSkipPerf.benchmark 1 thrpt 5 1430961236,046 ┬▒ 181925205,170 ops/s +/// StreamableSkipPerf.benchmark 10 thrpt 5 18226537,908 ┬▒ 671950,718 ops/s +/// StreamableSkipPerf.benchmark 100 thrpt 5 2395626,005 ┬▒ 88190,509 ops/s +/// StreamableSkipPerf.benchmark 1000 thrpt 5 215646,667 ┬▒ 2682,422 ops/s +/// StreamableSkipPerf.benchmark 10000 thrpt 5 21791,516 ┬▒ 1607,471 ops/s +/// StreamableSkipPerf.benchmark 100000 thrpt 5 2168,925 ┬▒ 32,591 ops/s +/// StreamableSkipPerf.benchmark 1000000 thrpt 5 214,055 ┬▒ 5,629 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 StreamableSkipPerf { + @Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" }) + public int times; + + Streamable result; + + @Setup + public void setup() { + result = Streamable.range(1, times).skip(times / 2); + } + + @Benchmark + public Object benchmark() { + return result.blockingLast(); + } +} diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 1b9f45e125..42f095fd6c 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -978,7 +978,7 @@ default Streamable retryWhen(BiFunction stage) { /** * Convenience method to await the completion of a stage, optimized - * for handling {@value #FINISHED} directly. + * for handling {@link #FINISHED} directly. * @param stage the stage to await */ static void awaitVoid(CompletionStage stage) { 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 bc26332fb3..c45204f3df 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 @@ -20,10 +20,10 @@ 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.exceptions.Exceptions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; -import io.reactivex.rxjava4.operators.IndexableSource; +import io.reactivex.rxjava4.operators.*; public record StreamableCollector( Streamable source, @@ -104,14 +104,59 @@ static final class CollectorStreamable } current = finisher.apply(storage); return NEXT_TRUE; + } else + if (upstream instanceof DeferredEnumerableSource dsrc) { + StreamableHelper.whenComplete(dsrc.enumerableReady(), this::deferredEnumerate); + return nextReady; + } else + if (upstream instanceof EnumerableSource esrc) { + try { + while (esrc.nextSync()) { + if (cancellation.isDisposed()) { + return CompletableFuture.failedFuture(new CancellationException()); + } + accumulator.accept(storage, (T)esrc.current()); + } + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedFuture(ex); + } + current = finisher.apply(storage); + return NEXT_TRUE; } - drain(); return nextReady; } return NEXT_FALSE; } + void deferredEnumerate(boolean hasInitialValue, Throwable error) { + if (error != null) { + nextReady.completeExceptionally(error); + return; + } + if (hasInitialValue) { + @SuppressWarnings("unchecked") + var upstreamCast = (DeferredEnumerableSource)upstream; + + try { + while (upstreamCast.nextSync()) { + if (cancellation.isDisposed()) { + nextReady.completeExceptionally(new CancellationException()); + return; + } + accumulator.accept(storage, upstreamCast.current()); + } + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + } + current = finisher.apply(storage); + nextReady.complete(true); + } + @Override public @NonNull R current() { return current; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java index 0ac897fde2..473cd833c9 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java @@ -14,13 +14,16 @@ package io.reactivex.rxjava4.internal.operators.streamable; import java.io.Serial; -import java.util.Iterator; +import java.util.*; 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.*; import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.Exceptions; public record StreamableConcatIterable( Iterable> sources, @@ -32,7 +35,8 @@ public record StreamableConcatIterable( return new ConcatIteratorStreamer<>(sources.iterator(), cancellation); } - static final class ConcatIteratorStreamer extends AtomicInteger implements Streamer { + static final class ConcatIteratorStreamer extends AtomicInteger + implements Streamer, BiConsumer { @Serial private static final long serialVersionUID = -9136569444189652718L; @@ -43,7 +47,7 @@ static final class ConcatIteratorStreamer extends AtomicInteger implements St DisposableStreamerCancellation currentCancellation; - Streamer upstream; + volatile Streamer upstream; CompletableFuture nextReady; @@ -55,9 +59,53 @@ static final class ConcatIteratorStreamer extends AtomicInteger implements St @Override public @NonNull CompletionStage next() { - nextReady = new CompletableFuture(); - drain(); - return nextReady; + for (;;) { + if (upstream == null) { + if (iterator.hasNext()) { + try { + var source = Objects.requireNonNull(iterator.next(), "The iterable returned a null Streamable"); + currentCancellation = cancellation.derive(); + upstream = source.stream(currentCancellation); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedFuture(ex); + } + } else { + return NEXT_FALSE; + } + } + var stage = upstream.next(); + if (stage instanceof CompletableFuture cf) { + if (cf.state() == State.SUCCESS) { + if (cf.getNow(false)) { + return NEXT_TRUE; + } else { + var finishStage = upstream.finish(); + if (finishStage instanceof CompletableFuture cff && cff.isDone()) { + upstream = null; + cancellation.delete(currentCancellation); + currentCancellation = null; + if (cff.isCompletedExceptionally()) { + return CompletableFuture.failedFuture(cff.exceptionNow()); + } + // will get the next source synchronously + } else { + nextReady = new CompletableFuture(); + finishStage.whenComplete(this::whenFinishComplete); + return nextReady; + } + } + } else { + nextReady = new CompletableFuture(); + stage.whenComplete(this); + return nextReady; + } + } else { + nextReady = new CompletableFuture(); + stage.whenComplete(this); + return nextReady; + } + } } @Override @@ -93,32 +141,37 @@ void drain() { nextReady.completeExceptionally(new NullPointerException("The iterator returned a null Streamable")); } else { upstream = nextStreamable.stream(currentCancellation); - drain(); } } else { nextReady.complete(false); } - } else { - upstream.next().whenComplete((v, e) -> { - if (e != null) { - nextReady.completeExceptionally(e); - } else - if (v) { - nextReady.complete(true); - } else { - cancellation.delete(currentCancellation); - upstream.finish().whenComplete((_, u) -> { - if (u != null) { - nextReady.completeExceptionally(u); - } else { - upstream = null; - drain(); - } - }); - } - }); + } + if (upstream != null) { + StreamableHelper.whenComplete(upstream.next(), this); } } while (decrementAndGet() != 0); } + + @Override + public void accept(Boolean v, Throwable e) { + if (e != null) { + nextReady.completeExceptionally(e); + } else + if (v) { + nextReady.complete(true); + } else { + cancellation.delete(currentCancellation); + StreamableHelper.whenComplete(upstream.finish(), this::whenFinishComplete); + } + } + + void whenFinishComplete(Void t, Throwable u) { + if (u != null) { + nextReady.completeExceptionally(u); + } else { + upstream = null; + drain(); + } + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArray.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArray.java index a0009cbfd3..06c51e2354 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArray.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArray.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 StreamableFromArray(@NonNull T[] items) implements Streamable { @@ -26,45 +27,67 @@ public record StreamableFromArray(@NonNull T[] items) implements Streamable(items); } - static final class FromArrayStreamer implements Streamer { + static final class FromArrayStreamer + implements Streamer, IndexableSource, EnumerableSource { final T[] items; - volatile int index; - - volatile T current; + int index; public FromArrayStreamer(T[] items) { this.items = items; + this.index = -1; } @Override public @NonNull CompletionStage next() { - var i = index; - if (i >= items.length) { + if (++index >= items.length) { return NEXT_FALSE; } - var nextItem = items[i]; - if (nextItem == null) { - index = items.length; - current = null; - return CompletableFuture.failedStage(new NullPointerException("Item at index " + i + " is null.")); + if (current() == null) { + return CompletableFuture.failedFuture(createNullError(index)); } - current = nextItem; - index = i + 1; return NEXT_TRUE; } + static NullPointerException createNullError(int index) { + return new NullPointerException("Item at index " + index + " is null."); + } + + @Override + public boolean nextSync() { + if (++index < items.length) { + if (current() == null) { + throw createNullError(index); + } + return true; + }; + return false; + } + @Override public @NonNull T current() { - return current; + return items[index]; } @Override public @NonNull CompletionStage finish() { index = items.length; - current = null; return FINISHED; } + + @Override + public @NonNull T elementAt(long index) throws Throwable { + var v = items[(int)index]; + if (v == null) { + throw createNullError((int)index); + } + return v; + } + + @Override + public long limit() { + return items.length; + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterable.java index 4bb2b294be..82591d8d79 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterable.java @@ -21,6 +21,7 @@ import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.internal.operators.streamable.StreamableEmpty.EmptyStreamer; +import io.reactivex.rxjava4.operators.EnumerableSource; public record StreamableFromIterable(@NonNull Iterable items) implements Streamable { @@ -40,13 +41,13 @@ public record StreamableFromIterable(@NonNull Iterable items) im return new IteratorStreamer<>(iterator); } - static final class IteratorStreamer implements Streamer { + static final class IteratorStreamer implements Streamer, EnumerableSource { Iterator iterator; long index; - volatile T current; + T current; IteratorStreamer(Iterator iterator) { this.iterator = iterator; @@ -58,7 +59,7 @@ static final class IteratorStreamer implements Streamer { var v = iterator.next(); current = v; if (v == null) { - return CompletableFuture.failedStage(new NullPointerException("Item at index " + index + " is null.")); + return CompletableFuture.failedStage(createNullError(index)); } index++; return NEXT_TRUE; @@ -67,6 +68,10 @@ static final class IteratorStreamer implements Streamer { return NEXT_FALSE; } + static NullPointerException createNullError(long index) { + return new NullPointerException("Item at index " + index + " is null."); + } + @Override public @NonNull T current() { return current; @@ -78,5 +83,20 @@ static final class IteratorStreamer implements Streamer { current = null; return FINISHED; } + + @Override + public boolean nextSync() throws Throwable { + if (index == 0L || iterator.hasNext()) { + var v = iterator.next(); + current = v; + if (v == null) { + throw createNullError(index); + } + index++; + return true; + } + current = null; + return false; + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java index 7d05680079..ecb876fedf 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java @@ -410,11 +410,11 @@ public static StreamableInterceptConfig createOnError(Consumer void whenComplete(CompletionStage stage, java.util.function.BiConsumer consumer) { - if (stage instanceof CompletableFuture cf && cf.isDone()) { - if (cf.isCompletedExceptionally()) { - consumer.accept(null, cf.exceptionNow()); - } else { - consumer.accept(cf.getNow(null), null); + if (stage instanceof CompletableFuture cf) { + switch (cf.state()) { + case SUCCESS -> consumer.accept(cf.getNow(null), null); + case CANCELLED, FAILED -> consumer.accept(null, cf.exceptionNow()); + default -> stage.whenComplete(consumer); } } else { stage.whenComplete(consumer); 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 be79e40f6d..b42620b635 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 @@ -86,14 +86,25 @@ void drain() { return; } + int wipMax = 1; + int wipIndex = 0; do { if (done) { - upstream.finish().whenComplete(this); + StreamableHelper.whenComplete(upstream.finish(), this); break; } else { - upstream.next().whenComplete(this); + StreamableHelper.whenComplete(upstream.next(), this); } - } while (decrementAndGet() != 0); + if (++wipIndex == wipMax) { + wipMax = get(); + if (wipIndex == wipMax) { + wipMax = addAndGet(-wipMax); + if (wipMax != 0) { + wipIndex = 0; + } + } + } + } while (wipMax != 0); } @Override 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 3f6c496761..aecd9f14f9 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 @@ -14,7 +14,8 @@ package io.reactivex.rxjava4.internal.operators.streamable; import java.util.Objects; -import java.util.concurrent.CompletionStage; +import java.util.concurrent.*; +import java.util.concurrent.Future.State; import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; @@ -23,29 +24,61 @@ import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; import io.reactivex.rxjava4.internal.util.ExceptionHelper; +import io.reactivex.rxjava4.operators.*; public record StreamableMap( @NonNull Streamable source, @NonNull Function mapper) implements Streamable, HasUpstreamStreamableSource { + @SuppressWarnings("unchecked") @Override public @NonNull Streamer<@NonNull R> stream(@NonNull StreamerCancellation cancellation) { - return new MapStreamer<>(source.stream(cancellation), mapper); + var upstream = source.stream(cancellation); + if (upstream instanceof IndexableSource isrc) { + return new MapStreamerIndexed<>(upstream, (IndexableSource)isrc, mapper); + } + if (upstream instanceof DeferredEnumerableSource esrc) { + return new MapStreamerDeferredEnumerated<>(upstream, (DeferredEnumerableSource)esrc, mapper); + } + if (upstream instanceof EnumerableSource esrc) { + return new MapStreamerEnumerated<>(upstream, (EnumerableSource)esrc, mapper); + } + return new MapStreamerBasic<>(upstream, mapper); } - static final class MapStreamer implements Streamer, java.util.function.Function { + static abstract class MapStreamerBase implements Streamer, java.util.function.Function { + final Streamer upstream; + final Function mapper; - volatile R current; - MapStreamer(Streamer upstream, Function mapper) { + + R current; + + MapStreamerBase(Streamer upstream, Function mapper) { this.upstream = upstream; this.mapper = mapper; } @Override public @NonNull CompletionStage next() { - return upstream.next().thenApply(this); + var stage = upstream.next(); + if (stage instanceof CompletableFuture cf) { + if (cf.state() == State.SUCCESS) { + if (cf.getNow(false)) { + try { + current = Objects.requireNonNull(mapper.apply(upstream.current()), "The mapper returned a null value"); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedFuture(ex); + } + return NEXT_TRUE; + } + return NEXT_FALSE; + } + } + + return stage.thenApply(this); } @Override @@ -73,4 +106,78 @@ public Boolean apply(Boolean e) { return upstream.finish(); } } + + static final class MapStreamerBasic extends MapStreamerBase { + MapStreamerBasic(Streamer upstream, Function mapper) { + super(upstream, mapper); + } + } + + static final class MapStreamerIndexed extends MapStreamerBase + implements IndexableSource { + + final IndexableSource indexed; + + MapStreamerIndexed(Streamer upstream, IndexableSource indexed, Function mapper) { + super(upstream, mapper); + this.indexed = indexed; + } + + @Override + public @NonNull R elementAt(long index) throws Throwable { + return Objects.requireNonNull(mapper.apply(indexed.elementAt(index)), "The mapper returned a null item"); + } + + @Override + public long limit() { + return indexed.limit(); + } + } + + static final class MapStreamerEnumerated extends MapStreamerBase + implements EnumerableSource { + + final EnumerableSource enumerable; + + MapStreamerEnumerated(Streamer upstream, EnumerableSource enumerable, Function mapper) { + super(upstream, mapper); + this.enumerable = enumerable; + } + + @Override + public boolean nextSync() throws Throwable { + if (enumerable.nextSync()) { + current = Objects.requireNonNull(mapper.apply(enumerable.current()), "The mapper returned a null item"); + return true; + } + return false; + } + + } + + static final class MapStreamerDeferredEnumerated extends MapStreamerBase + implements DeferredEnumerableSource { + + final DeferredEnumerableSource enumerable; + + MapStreamerDeferredEnumerated(Streamer upstream, DeferredEnumerableSource enumerable, Function mapper) { + super(upstream, mapper); + this.enumerable = enumerable; + } + + @Override + public boolean nextSync() throws Throwable { + if (enumerable.nextSync()) { + current = Objects.requireNonNull(mapper.apply(enumerable.current()), "The mapper returned a null item"); + return true; + } + return false; + } + + @Override + public CompletionStage enumerableReady() { + return enumerable.enumerableReady(); + } + + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java index 2e525960a9..720921d74e 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java @@ -25,6 +25,7 @@ import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.disposables.*; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamSingleSource; +import io.reactivex.rxjava4.operators.DeferredEnumerableSource; public record StreamableSingleFlattenAs( SingleSource source, @@ -41,7 +42,8 @@ public record StreamableSingleFlattenAs( static final class FlattenAsSingleObserver extends AtomicInteger - implements SingleObserver, Streamer, DisposableOnly, java.util.function.Function { + implements SingleObserver, Streamer, DisposableOnly, + DeferredEnumerableSource { @Serial private static final long serialVersionUID = 796267562672678347L; @@ -54,11 +56,11 @@ static final class FlattenAsSingleObserver final CompletableFuture iteratorReady; - volatile Iterator iteratorHandover; - U current; - Iterator currentIterator; + volatile Iterator currentIterator; + + boolean deferredOnce; FlattenAsSingleObserver( Function> mapper, @@ -73,7 +75,11 @@ static final class FlattenAsSingleObserver public @NonNull CompletionStage next() { var it = currentIterator; if (it == null) { - return iteratorReady.thenApply(this); + return iteratorReady; + } + if (!deferredOnce) { + deferredOnce = true; + return NEXT_TRUE; } if (it.hasNext()) { current = it.next(); @@ -82,19 +88,6 @@ static final class FlattenAsSingleObserver return NEXT_FALSE; } - @Override - public Boolean apply(Boolean t) { - if (t) { - currentIterator = iteratorHandover; - iteratorHandover = null; - return true; - } else { - currentIterator = null; - iteratorHandover = null; - } - return false; - } - @Override public @NonNull U current() { return current; @@ -122,9 +115,9 @@ public void onSubscribe(@NonNull Disposable d) { public void onSuccess(@NonNull T t) { try { var iterator = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null Iterable").iterator(); - this.iteratorHandover = iterator; if (iterator.hasNext()) { current = iterator.next(); + this.currentIterator = iterator; iteratorReady.complete(true); } else { iteratorReady.complete(false); @@ -142,5 +135,24 @@ public void onError(@NonNull Throwable e) { upstream.lazySet(DisposableHelper.DISPOSED); iteratorReady.completeExceptionally(e); } + + @Override + public CompletionStage enumerableReady() { + return iteratorReady; + } + + @Override + public boolean nextSync() throws Throwable { + // because onSuccess will pull out the first item + if (!deferredOnce) { + deferredOnce = true; + return true; + } + if (currentIterator.hasNext()) { + current = currentIterator.next(); + return true; + } + return false; + } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java index a2fd7f0cb0..95e41fe2e5 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java @@ -59,9 +59,20 @@ void drain() { if (getAndIncrement() != 0) { return; } + int wipMax = 1; + int wipIndex = 0; do { - upstream.next().whenComplete(this); - } while (decrementAndGet() != 0); + StreamableHelper.whenComplete(upstream.next(), this); + if (++wipIndex == wipMax) { + wipMax = get(); + if (wipIndex == wipMax) { + wipMax = addAndGet(-wipMax); + if (wipMax != 0) { + wipIndex = 0; + } + } + } + } while (wipMax != 0); } @Override diff --git a/src/main/java/io/reactivex/rxjava4/operators/DeferredEnumerableSource.java b/src/main/java/io/reactivex/rxjava4/operators/DeferredEnumerableSource.java new file mode 100644 index 0000000000..a84a86a8ad --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/operators/DeferredEnumerableSource.java @@ -0,0 +1,36 @@ +/* + * 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.operators; + +import java.util.concurrent.CompletionStage; + +/// Represents an unknown length, deferred iterable source which can be moved forward synchronously +/// and obtain the current item via a simple call once it reports said iterable elements are ready +/// to be consumed. +///

+/// No {@code hasNext} and {@code next} duplication. C# IEnumerator is way better in this regard. +/// @param the element type of the source +/// @see IndexableSource +/// @since 4.0.0 +public interface DeferredEnumerableSource extends EnumerableSource { + + /** + * Returns true if the source is ready to be consumed via its + * {@link EnumerableSource#nextSync()} and {@link EnumerableSource#current()} + * methods. + * @return the completion stage that indicates an empty {@code false} or a non-empty + * {@code true} enumerable source is now available + */ + CompletionStage enumerableReady(); +} diff --git a/src/main/java/io/reactivex/rxjava4/operators/EnumerableSource.java b/src/main/java/io/reactivex/rxjava4/operators/EnumerableSource.java new file mode 100644 index 0000000000..d42a3d7193 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/operators/EnumerableSource.java @@ -0,0 +1,41 @@ +/* + * 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.operators; + +/// Represents an unknown length iterable source which can be moved forward synchronously and obtain the current +/// item via a simple call. +///

+/// No {@code hasNext} and {@code next} duplication. C# IEnumerator is way better in this regard. +/// @param the element type of the source +/// @see IndexableSource +/// @since 4.0.0 +public interface EnumerableSource { + + /** + * Synchronously obtains the next item or returns {@code false} if no + * more items. + * @return {@code true} it there is an item available which can be obtained via {@link #current()}, + * {@code false} if no more items are available + * @throws Throwable if there is a (processing) error while going to the next item synchronously + */ + boolean nextSync() throws Throwable; + + /** + * Returns the current item if {@link #nextSync()} returned {@code true} the previous call. + *

+ * Calling before the first or after exhaustion of the source is an undefined behavior + * @return the current item + */ + T current(); // FIXME not sure about the name clash with Streamable.current +} 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 bf6d48bb23..a543476a47 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 @@ -22,6 +22,7 @@ import io.reactivex.rxjava4.core.Streamable; import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableConcatIterableTest extends StreamableBaseTest { @@ -37,6 +38,42 @@ public void normal() throws Throwable { ); } + @Test + public void normalHidden() throws Throwable { + Streamable.concat(List.of(Streamable.range(1, 5).hide(), Streamable.range(6, 5).hide(), Streamable.range(11, 5).hide())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult( + 1, 2, 3, 4, 5, + 6, 7, 8, 9, 10, + 11, 12, 13, 14, 15 + ); + } + + @Test + public void normalMixedHidden1() throws Throwable { + Streamable.concat(List.of(Streamable.range(1, 5).hide(), Streamable.range(6, 5), Streamable.range(11, 5).hide())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult( + 1, 2, 3, 4, 5, + 6, 7, 8, 9, 10, + 11, 12, 13, 14, 15 + ); + } + + @Test + public void normalHidden2() throws Throwable { + Streamable.concat(List.of(Streamable.range(1, 5), Streamable.range(6, 5).hide(), 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 + ); + } + @Test public void crash() throws Throwable { Streamable.concat(List.of(Streamable.range(1, 5), Streamable.error(new TestException()), Streamable.range(11, 5))) @@ -121,4 +158,150 @@ public void lotEmpties() { .awaitDone(5, TimeUnit.SECONDS) .assertResult(); } + + @Test + public void delayed() 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 + ); + } + + @Test + public void intervalRange() throws Throwable { + Streamable.concat(List.of( + Streamable.intervalRange(1, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()), + Streamable.intervalRange(6, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()), + Streamable.intervalRange(11, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()))) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult( + 1L, 2L, 3L, 4L, 5L, + 6L, 7L, 8L, 9L, 10L, + 11L, 12L, 13L, 14L, 15L + ); + } + + @Test + public void virtualCreate() { + Streamable.concat(List.of(Streamable.empty(), + Streamable.create(emitter -> { + emitter.emit(1); + emitter.emit(2); + emitter.emit(3); + emitter.emit(4); + emitter.emit(5); + }), + Streamable.create(_ -> { + }) + )) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void virtualCreate2() { + Streamable.concat(List.of(Streamable.empty(), + Streamable.create(emitter -> { + emitter.emit(1); + emitter.emit(2); + emitter.emit(3); + emitter.emit(4); + emitter.emit(5); + }), + Streamable.create(emitter -> { + emitter.emit(6); + emitter.emit(7); + }) + )) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5, 6, 7); + } + + @Test + public void virtualCreateNull() { + Streamable.concat(Arrays.asList(Streamable.empty(), + Streamable.create(emitter -> { + emitter.emit(1); + emitter.emit(2); + emitter.emit(3); + emitter.emit(4); + emitter.emit(5); + }), + Streamable.create(_ -> { + }), + null + )) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class, 1, 2, 3, 4, 5); + } + + @Test + public void virtualCreateNullDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.concat(Arrays.asList(Streamable.empty(), + Streamable.create(emitter -> { + emitter.emit(1); + emitter.emit(2); + emitter.emit(3); + emitter.emit(4); + emitter.emit(5); + }, exec), + Streamable.create(_ -> { + }, exec), + null + )) + .test(exec) + .awaitDone(500, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class, 1, 2, 3, 4, 5); + }); + } + + @Test + public void virtualCreateError() { + Streamable.concat(Arrays.asList(Streamable.empty(), + Streamable.create(emitter -> { + emitter.emit(1); + emitter.emit(2); + emitter.emit(3); + emitter.emit(4); + emitter.emit(5); + }), + Streamable.create(_ -> { + }), + Streamable.error(new TestException()) + )) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 1, 2, 3, 4, 5); + } + + @Test + public void virtualCreateFinishFail() { + Streamable.concat(Arrays.asList(Streamable.empty(), + Streamable.create(emitter -> { + emitter.emit(1); + emitter.emit(2); + emitter.emit(3); + emitter.emit(4); + emitter.emit(5); + }), + Streamable.create(_ -> { + }), + StreamableFailingFinish.MAIN_COMPLETES + )) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 1, 2, 3, 4, 5); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArrayTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArrayTest.java index 96ceb72522..a3a5642fa8 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArrayTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromArrayTest.java @@ -13,9 +13,12 @@ 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 StreamableFromArrayTest extends StreamableBaseTest { @@ -38,4 +41,21 @@ public void hasNull() throws Throwable { ; } + @Test + public void indexable() { + Streamable.fromArray(1, 2, 3) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3)); + } + + @Test + public void indexableNull() { + Streamable.fromArray(1, null, 3) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterableTest.java index 25243761f0..c94a0e14a0 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterableTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromIterableTest.java @@ -15,15 +15,18 @@ 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; +import io.reactivex.rxjava4.exceptions.TestException; public class StreamableFromIterableTest extends StreamableBaseTest { @Test public void normal() throws Throwable { - Streamable.fromStream(List.of(1, 2, 3).stream()) + Streamable.fromIterable(List.of(1, 2, 3)) .test() .awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3); @@ -31,7 +34,7 @@ public void normal() throws Throwable { @Test public void empty() throws Throwable { - Streamable.fromStream(List.of().stream()) + Streamable.fromIterable(List.of()) .test() .awaitDone(5, TimeUnit.SECONDS) .assertResult(); @@ -39,7 +42,7 @@ public void empty() throws Throwable { @Test public void one() throws Throwable { - Streamable.fromStream(List.of(1).stream()) + Streamable.fromIterable(List.of(1)) .test() .awaitDone(5, TimeUnit.SECONDS) .assertResult(1); @@ -47,7 +50,7 @@ public void one() throws Throwable { @Test public void hasNull() throws Throwable { - Streamable.fromStream(Arrays.asList(1, null, 3).stream()) + Streamable.fromIterable(Arrays.asList(1, null, 3)) .test() .awaitDone(5, TimeUnit.SECONDS) .assertFailure(NullPointerException.class, 1) @@ -57,11 +60,39 @@ public void hasNull() throws Throwable { @Test public void hasNull2() throws Throwable { - Streamable.fromStream(Arrays.asList(null, 1, 2, 3).stream()) + Streamable.fromIterable(Arrays.asList(null, 1, 2, 3)) .test() .awaitDone(5, TimeUnit.SECONDS) .assertFailure(NullPointerException.class) .assertError(t -> t.getMessage().equals("Item at index 0 is null.")); ; } + + @Test + public void iteratorThrows() throws Throwable { + Streamable.fromIterable(() -> { throw new TestException("test"); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class) + .assertError(t -> t.getMessage().equals("test")); + ; + } + + @Test + public void enumerableSource() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void enumerableSourceWithNull() { + Streamable.fromIterable(Arrays.asList(1, 2, null, 4, 5)) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromStreamTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromStreamTest.java index af08e5a211..5b01348178 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromStreamTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromStreamTest.java @@ -24,7 +24,7 @@ public class StreamableFromStreamTest extends StreamableBaseTest { @Test public void normal() throws Throwable { - Streamable.fromIterable(List.of(1, 2, 3)) + Streamable.fromStream(List.of(1, 2, 3).stream()) .test() .awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3); @@ -32,7 +32,7 @@ public void normal() throws Throwable { @Test public void empty() throws Throwable { - Streamable.fromIterable(List.of()) + Streamable.fromStream(List.of().stream()) .test() .awaitDone(5, TimeUnit.SECONDS) .assertResult(); @@ -40,7 +40,7 @@ public void empty() throws Throwable { @Test public void one() throws Throwable { - Streamable.fromIterable(List.of(1)) + Streamable.fromStream(List.of(1).stream()) .test() .awaitDone(5, TimeUnit.SECONDS) .assertResult(1); @@ -48,7 +48,7 @@ public void one() throws Throwable { @Test public void hasNull() throws Throwable { - Streamable.fromIterable(Arrays.asList(1, null, 3)) + Streamable.fromStream(Arrays.asList(1, null, 3).stream()) .test() .awaitDone(5, TimeUnit.SECONDS) .assertFailure(NullPointerException.class, 1) @@ -58,21 +58,11 @@ public void hasNull() throws Throwable { @Test public void hasNull2() throws Throwable { - Streamable.fromIterable(Arrays.asList(null, 1, 2, 3)) + Streamable.fromStream(Arrays.asList(null, 1, 2, 3).stream()) .test() .awaitDone(5, TimeUnit.SECONDS) .assertFailure(NullPointerException.class) .assertError(t -> t.getMessage().equals("Item at index 0 is null.")); ; } - - @Test - public void iteratorThrows() throws Throwable { - Streamable.fromIterable(() -> { throw new TestException("test"); }) - .test() - .awaitDone(5, TimeUnit.SECONDS) - .assertFailure(TestException.class) - .assertError(t -> t.getMessage().equals("test")); - ; - } } 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 ac4617efc7..11bf6d4414 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 @@ -13,11 +13,16 @@ package io.reactivex.rxjava4.internal.operators.streamable; +import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.stream.*; 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.processors.DispatchStreamProcessor; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableMapTest extends StreamableBaseTest { @@ -30,6 +35,16 @@ public void basic() { .assertResult("1", "2", "3", "4", "5"); } + @Test + public void basicHidden() { + Streamable.range(1, 5) + .hide() + .map(v -> v.toString()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("1", "2", "3", "4", "5"); + } + @Test public void mapperNull() { Streamable.range(1, 5) @@ -39,6 +54,15 @@ public void mapperNull() { .assertFailure(NullPointerException.class); } + @Test + public void error() { + Streamable.error(new TestException()) + .map(v -> v.toString()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + @Test public void mapperCrash() { Streamable.range(1, 5) @@ -48,4 +72,145 @@ public void mapperCrash() { .assertFailure(TestException.class); } + @Test + public void delayed() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .map(v -> v.toString()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("1", "2", "3", "4", "5"); + } + + @Test + public void delayedEmpty() { + Streamable.empty() + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .map(v -> v.toString()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void delayedMapperNull() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .map(_ -> null) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void delayedMapperCrash() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .map(_ -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void complete() throws Throwable { + var dsp = new DispatchStreamProcessor(); + + var ts = dsp.map(v -> v.toString()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.finish(null); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void indexed() { + Streamable.range(1, 5) + .map(v -> v.toString()) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("1", "2", "3", "4", "5")); + } + + @Test + public void indexedNull() { + Streamable.range(1, 5) + .map(_ -> null) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void indexedCrash() { + Streamable.range(1, 5) + .map(_ -> { throw new TestException(); }) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void enumerated() { + Streamable.fromIterable(() -> IntStream.range(1, 6).iterator()) + .map(v -> v.toString()) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("1", "2", "3", "4", "5")); + } + + @Test + public void enumeratedNull() { + Streamable.fromIterable(() -> IntStream.range(1, 6).iterator()) + .map(_ -> null) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void enumeratedCrash() { + Streamable.fromIterable(() -> IntStream.range(1, 6).iterator()) + .map(_ -> { throw new TestException(); }) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void deferredEnumerable() throws Throwable { + Single.just(List.of(1, 2, 3, 4, 5)) + .flattenAsStreamable(v -> v) + .map(v -> v + 1) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(2, 3, 4, 5, 6)); + } + + @Test + public void deferredEnumerableDebug() throws Throwable { + withCachedExecutor(exec -> { + Single.just(List.of(1, 2, 3, 4, 5)) + .flattenAsStreamable(v -> v) + .map(v -> v + 1) + .collect(Collectors.toList()) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(2, 3, 4, 5, 6)); + }); + } + } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java index 3feae44e35..a39bdc3767 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java @@ -15,6 +15,7 @@ import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; @@ -33,6 +34,17 @@ public void normal() throws Throwable { .assertResult(10, 20, 30, 40, 50); } + @Test + public void normalDebug() throws Throwable { + withCachedExecutor(exec -> { + Single.just(1) + .flattenAsStreamable(v -> List.of(v * 10, v * 20, v * 30, v * 40, v * 50)) + .test(exec) + .awaitDone(500, TimeUnit.SECONDS) + .assertResult(10, 20, 30, 40, 50); + }); + } + @Test public void error() throws Throwable { Single.error(new TestException()) @@ -84,4 +96,14 @@ public void cancel() throws Throwable { awaitCondition(false, () -> ss.hasObservers(), 1000); } + + @Test + public void deferredEnumerable() throws Throwable { + Single.just(List.of(1, 2, 3, 4, 5)) + .flattenAsStreamable(v -> v) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } }