diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java index 1f617c559d..07ed004062 100644 --- a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java @@ -13,11 +13,13 @@ package io.reactivex.rxjava4.streamable; +import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.openjdk.jmh.annotations.*; -import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.Streamable; /// /// The map is one of the most used operator in the ecosystem so it must be fast and @@ -66,6 +68,48 @@ /// StreamableSkipPerf.benchmark 100000 thrpt 5 2168,925 ┬▒ 32,591 ops/s /// StreamableSkipPerf.benchmark 1000000 thrpt 5 214,055 ┬▒ 5,629 ops/s /// ``` +/// +/// # 3. sync bias via Claude Fable atomics +/// +/// Small regression on 1, but +200% on million vs 2., almost +410% vs baseline +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableSkipPerf.benchmark 1 thrpt 5 1392748873,097 ┬▒ 240564597,648 ops/s +/// StreamableSkipPerf.benchmark 10 thrpt 5 50478482,677 ┬▒ 2503440,823 ops/s +/// StreamableSkipPerf.benchmark 100 thrpt 5 8260512,961 ┬▒ 832256,445 ops/s +/// StreamableSkipPerf.benchmark 1000 thrpt 5 617903,033 ┬▒ 127135,231 ops/s +/// StreamableSkipPerf.benchmark 10000 thrpt 5 65116,345 ┬▒ 8425,364 ops/s +/// StreamableSkipPerf.benchmark 100000 thrpt 5 6298,539 ┬▒ 739,686 ops/s +/// StreamableSkipPerf.benchmark 1000000 thrpt 5 634,945 ┬▒ 99,134 ops/s +/// ``` +/// +/// # 4. indexable/enumerable/deferredenumerable +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableSkipPerf.benchmark 1 thrpt 5 1371065918,333 ┬▒ 114154178,555 ops/s +/// StreamableSkipPerf.benchmark 10 thrpt 5 48387844,296 ┬▒ 1998160,368 ops/s +/// StreamableSkipPerf.benchmark 100 thrpt 5 8344391,259 ┬▒ 251723,213 ops/s +/// StreamableSkipPerf.benchmark 1000 thrpt 5 700332,413 ┬▒ 28667,777 ops/s +/// StreamableSkipPerf.benchmark 10000 thrpt 5 68645,229 ┬▒ 3123,103 ops/s +/// StreamableSkipPerf.benchmark 100000 thrpt 5 6047,152 ┬▒ 725,594 ops/s +/// StreamableSkipPerf.benchmark 1000000 thrpt 5 649,134 ┬▒ 36,349 ops/s +/// StreamableSkipPerf.enumerable 1 thrpt 5 71117553,552 ┬▒ 6738602,495 ops/s +/// StreamableSkipPerf.enumerable 10 thrpt 5 43426136,653 ┬▒ 2744303,836 ops/s +/// StreamableSkipPerf.enumerable 100 thrpt 5 2819347,969 ┬▒ 20437,839 ops/s +/// StreamableSkipPerf.enumerable 1000 thrpt 5 298000,822 ┬▒ 28754,714 ops/s +/// StreamableSkipPerf.enumerable 10000 thrpt 5 28839,499 ┬▒ 3096,840 ops/s +/// StreamableSkipPerf.enumerable 100000 thrpt 5 2657,147 ┬▒ 23,023 ops/s +/// StreamableSkipPerf.enumerable 1000000 thrpt 5 187,770 ┬▒ 42,082 ops/s +/// StreamableSkipPerf.indexed 1 thrpt 5 75403131,814 ┬▒ 4676965,390 ops/s +/// StreamableSkipPerf.indexed 10 thrpt 5 42730509,975 ┬▒ 6004178,786 ops/s +/// StreamableSkipPerf.indexed 100 thrpt 5 3561284,547 ┬▒ 381609,301 ops/s +/// StreamableSkipPerf.indexed 1000 thrpt 5 346781,462 ┬▒ 28909,008 ops/s +/// StreamableSkipPerf.indexed 10000 thrpt 5 54254,177 ┬▒ 7440,753 ops/s +/// StreamableSkipPerf.indexed 100000 thrpt 5 5153,256 ┬▒ 310,895 ops/s +/// StreamableSkipPerf.indexed 1000000 thrpt 5 259,912 ┬▒ 76,746 ops/s +/// ``` @BenchmarkMode(Mode.Throughput) @Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) @Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) @@ -77,14 +121,28 @@ public class StreamableSkipPerf { public int times; Streamable result; + Streamable> resultIndexed; + Streamable> resultEnumerable; @Setup public void setup() { result = Streamable.range(1, times).skip(times / 2); + resultIndexed = Streamable.range(1, times).skip(times / 2).collect(Collectors.toList()); + resultEnumerable = Streamable.range(1, times).filter(_ -> true).skip(times / 2).collect(Collectors.toList()); } @Benchmark public Object benchmark() { return result.blockingLast(); } + + @Benchmark + public Object indexed() { + return resultIndexed.blockingLast(); + } + + @Benchmark + public Object enumerable() { + return resultEnumerable.blockingLast(); + } } diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 42f095fd6c..8b7073602a 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -623,6 +623,21 @@ static Streamable> zip(Iterable> s default T blockingFirst() { return StreamableBlocking.blockingFirst(this); } + /** + * Blocks the current thread until this {@code Streamable} produces one item, which is then returned. + * @param cancellation the external cancellation resource to pass into the chain + * @return the first item of this {@code Streamable} + * @throws NoSuchElementException if the this {@code Streamable} is empty + * @throws CancellationException if this {@code Streamable} failed with a checked exception + * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception + * @throws NullPointerException if {@code cancellation} is {@code null} + */ + @CheckReturnValue + @NonNull + default T blockingFirst(StreamerCancellation cancellation) { + Objects.requireNonNull(cancellation, "cancellation is null"); + return StreamableBlocking.blockingFirst(this, cancellation); + } /** * Blocks the current thread until this {@code Streamable} produces all of its items @@ -638,6 +653,23 @@ default T blockingLast() { return StreamableBlocking.blockingLast(this); } + /** + * Blocks the current thread until this {@code Streamable} produces all of its items + * and the very last is then returned. + * @param cancellation the external cancellation resource to pass into the chain + * @return the very last item of this {@code Streamable} + * @throws NoSuchElementException if the this {@code Streamable} is empty + * @throws CancellationException if this {@code Streamable} failed with a checked exception + * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception + * @throws NullPointerException if {@code cancellation} is {@code null} + */ + @CheckReturnValue + @NonNull + default T blockingLast(StreamerCancellation cancellation) { + Objects.requireNonNull(cancellation, "cancellation is null"); + return StreamableBlocking.blockingLast(this, cancellation); + } + /** * Collects all upstream values via the use of a {@link Collector} configuration * and emits its resulting value as a single item of the returned {@code Streamable}. diff --git a/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java b/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java index 65910a0726..50713a1fdb 100644 --- a/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java +++ b/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java @@ -13,6 +13,8 @@ package io.reactivex.rxjava4.disposables; +import io.reactivex.rxjava4.internal.disposables.NeverDisposableStreamerCancellation; + /** * Represents the full, disposable cancellation interface for {@code Streamer} * operations. @@ -23,4 +25,13 @@ */ public interface DisposableStreamerCancellation extends StreamerCancellation, Disposable { + /** + * Returns a constant instance which does nothing, cannot be disposed and + * accepts any incoming Disposable without registering it or handling it in any form, + * because this {@code never} instance cannot be disposed to begin with. + * @return the shared constant no-op instance + */ + static DisposableStreamerCancellation never() { + return NeverDisposableStreamerCancellation.INSTANCE; + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/disposables/NeverDisposableStreamerCancellation.java b/src/main/java/io/reactivex/rxjava4/internal/disposables/NeverDisposableStreamerCancellation.java new file mode 100644 index 0000000000..33a059af1c --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/disposables/NeverDisposableStreamerCancellation.java @@ -0,0 +1,54 @@ +/* + * 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.disposables; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.disposables.*; + +/// A [DisposableStreamerCancellation] handler that does nothing. +public enum NeverDisposableStreamerCancellation implements DisposableStreamerCancellation { + INSTANCE + ; + + @Override + public boolean isDisposed() { + return false; // always because it has no state + } + + @Override + public boolean add(@NonNull Disposable d) { + return true; // always succeeds to avoid infinite retry loops + } + + @Override + public boolean remove(@NonNull Disposable d) { + return true; // always succeeds + } + + @Override + public boolean delete(@NonNull Disposable d) { + return true; // always succeeds + } + + @Override + public @NonNull DisposableStreamerCancellation derive() { + return this; + } + + @Override + public void dispose() { + // deliberately no-op + } + +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsEnumerableStreamable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsEnumerableStreamable.java new file mode 100644 index 0000000000..24cd5003a3 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsEnumerableStreamable.java @@ -0,0 +1,26 @@ +/* + * 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 io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.operators.EnumerableSource; + +/// Marker interface to indicate a [Streamable] source will produce +/// an [EnumerableSource]-enabled [Streamer] and thus enables +/// optimizations and operator fusion during assembly time. +/// @param the element type of the `Streamable` +/// @since 4.0.0 +public interface IsEnumerableStreamable extends Streamable { + +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsIndexableStreamable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsIndexableStreamable.java new file mode 100644 index 0000000000..498378704c --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsIndexableStreamable.java @@ -0,0 +1,26 @@ +/* + * 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 io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.operators.IndexableSource; + +/// Marker interface to indicate a [Streamable] source will produce +/// an [IndexableSource]-enabled [Streamer] and thus enables +/// optimizations and operator fusion during assembly time. +/// @param the element type of the `Streamable` +/// @since 4.0.0 +public interface IsIndexableStreamable extends Streamable { + +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsSynchronousStreamable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsSynchronousStreamable.java new file mode 100644 index 0000000000..adee8eccf5 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsSynchronousStreamable.java @@ -0,0 +1,26 @@ +/* + * 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 io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.operators.*; + +/// Marker interface to indicate a [Streamable] source will produce +/// both an [IndexableSource] and an [EnumerableSource] capable +/// [Streamer]. +/// @param the element type of the `Streamable` +/// @since 4.0.0 +public interface IsSynchronousStreamable extends IsIndexableStreamable, IsEnumerableStreamable { + +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java index 9b75fd840e..cf2e1583fe 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java @@ -18,7 +18,7 @@ import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.core.Streamable; -import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.internal.util.ExceptionHelper; @@ -36,7 +36,23 @@ public record StreamableBlocking() { @CheckReturnValue @NonNull public static T blockingFirst(Streamable source) { - var streamer = source.stream(new CompositeDisposable()); + return blockingFirst(source, new CompositeDisposable()); + } + + /** + * Consumes the first item and finishes the {@link Streamable}, + * throwing {@link NoSuchElementException} if the source is empty. + * @param the element type + * @param source the source {@code Streamable} + * @param cancellation the external cancellation manager + * @return the first item + * @throws RuntimeException if the source signals an unchecked exception + * @throws CompletionException if the source signals a checked exception + */ + @CheckReturnValue + @NonNull + public static T blockingFirst(Streamable source, StreamerCancellation cancellation) { + var streamer = source.stream(cancellation); Throwable nextException = null; Throwable finishException = null; T result = null; @@ -74,7 +90,21 @@ public static T blockingFirst(Streamable source) { * @throws CompletionException if the source signals a checked exception */ public static T blockingLast(Streamable source) { - var streamer = source.stream(new CompositeDisposable()); + return blockingLast(source, new CompositeDisposable()); + } + + /** + * Consumes all upstream items and returns the very last or throws + * a {@link NoSuchElementException}. + * @param the element type + * @param source the source sequence + * @param cancellation the external cancellation manager + * @return the very last value + * @throws RuntimeException if the source signals an unchecked exception + * @throws CompletionException if the source signals a checked exception + */ + public static T blockingLast(Streamable source, StreamerCancellation cancellation) { + var streamer = source.stream(cancellation); Throwable nextException = null; Throwable finishException = null; T result = null; 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 4083d862bd..dcb4c6fb6c 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 @@ -21,7 +21,7 @@ import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.operators.*; -public enum StreamableEmpty implements Streamable { +public enum StreamableEmpty implements IsSynchronousStreamable { INSTANCE; 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 2bf50f9e2e..43b570a031 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 @@ -20,7 +20,7 @@ import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.operators.*; -public record StreamableJust(@NonNull T item) implements Streamable { +public record StreamableJust(@NonNull T item) implements IsSynchronousStreamable { @Override public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { 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 43f0e6783a..5ca9e47638 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 @@ -20,7 +20,7 @@ import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.operators.*; -public record StreamableRange(int start, int count) implements Streamable { +public record StreamableRange(int start, int count) implements IsSynchronousStreamable { @Override public @NonNull Streamer<@NonNull Integer> stream(@NonNull StreamerCancellation cancellation) { 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 eec67d1921..1f58bb32a7 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 @@ -20,7 +20,7 @@ import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.operators.*; -public record StreamableRangeLong(long start, long count) implements Streamable { +public record StreamableRangeLong(long start, long count) implements IsSynchronousStreamable { @Override public @NonNull Streamer<@NonNull Long> stream(@NonNull StreamerCancellation cancellation) { 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 95e41fe2e5..adfeaaf7a6 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 @@ -15,21 +15,34 @@ 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.*; import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.operators.*; public record StreamableSkip(Streamable source, long count) implements Streamable { + @SuppressWarnings("unchecked") @Override public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { - return new SkipStreamer<>(source.stream(cancellation), count); + var upstream = source.stream(cancellation); + if (upstream instanceof IndexableSource isrc) { + return new SkipStreamerIndexed<>(upstream, (IndexableSource)isrc, count); + } else + if (upstream instanceof DeferredEnumerableSource dsrc) { + return new SkipStreamerDeferredEnumerable<>(upstream, (DeferredEnumerableSource)dsrc, count); + } else + if (upstream instanceof EnumerableSource esrc) { + return new SkipStreamerEnumerable<>(upstream, (EnumerableSource)esrc, count); + } + return new SkipStreamerBasic<>(upstream, count); } - static final class SkipStreamer extends AtomicInteger implements Streamer, BiConsumer { + static abstract class SkipStreamer extends AtomicInteger implements Streamer, BiConsumer { @Serial private static final long serialVersionUID = 1988154737845167665L; @@ -56,38 +69,52 @@ static final class SkipStreamer extends AtomicInteger implements Streamer, } void drain() { - if (getAndIncrement() != 0) { - return; - } - int wipMax = 1; - int wipIndex = 0; - do { - StreamableHelper.whenComplete(upstream.next(), this); - if (++wipIndex == wipMax) { - wipMax = get(); - if (wipIndex == wipMax) { - wipMax = addAndGet(-wipMax); - if (wipMax != 0) { - wipIndex = 0; + for (;;) { + var nextStage = upstream.next().toCompletableFuture(); + var state = nextStage.state(); + if (state == State.RUNNING) { + set(1); + nextStage.whenComplete(this); + if (compareAndSet(1, 0)) { + return; + } + state = nextStage.state(); + } + + if (state == State.SUCCESS) { + if (nextStage.getNow(false)) { + if (remaining-- <= 0) { + waiter.complete(true); + return; } + // still skipping, try the next upstream value + } else { + waiter.complete(false); + return; } + } else { + waiter.completeExceptionally(nextStage.exceptionNow()); + return; } - } while (wipMax != 0); + } } @Override public void accept(Boolean t, Throwable u) { - if (u != null) { - waiter.completeExceptionally(u); - } else - if (t) { - if (remaining-- > 0) { - drain(); + if (!compareAndSet(1, 2)) { + if (u != null) { + waiter.completeExceptionally(u); } else { - waiter.complete(true); + if (t) { + if (remaining-- <= 0) { + waiter.complete(true); + } else { + drain(); + } + } else { + waiter.complete(false); + } } - } else { - waiter.complete(false); } } @@ -101,4 +128,96 @@ public void accept(Boolean t, Throwable u) { return upstream.finish(); } } + + static final class SkipStreamerBasic extends SkipStreamer { + + @Serial + private static final long serialVersionUID = 7002195716600087893L; + + SkipStreamerBasic(Streamer upstream, long count) { + super(upstream, count); + } + } + + static final class SkipStreamerIndexed extends SkipStreamer + implements IndexableSource { + + @Serial + private static final long serialVersionUID = 773832461044750722L; + + final IndexableSource indexable; + + final long count; + + SkipStreamerIndexed(Streamer upstream, IndexableSource indexable, long count) { + super(upstream, count); + this.indexable = indexable; + this.count = count; + } + + @Override + public @NonNull T elementAt(long index) throws Throwable { + return indexable.elementAt(index + count); + } + + @Override + public long limit() { + return Math.max(0, indexable.limit() - count); + } + } + + static final class SkipStreamerEnumerable extends SkipStreamer + implements EnumerableSource { + + @Serial + private static final long serialVersionUID = 773832461044750722L; + + final EnumerableSource enumerable; + + SkipStreamerEnumerable(Streamer upstream, EnumerableSource enumerable, long count) { + super(upstream, count); + this.enumerable = enumerable; + } + + @Override + public boolean nextSync() throws Throwable { + while (remaining-- > 0) { + if (!enumerable.nextSync()) { + return false; + } + } + return enumerable.nextSync(); + } + + } + + static final class SkipStreamerDeferredEnumerable extends SkipStreamer + implements DeferredEnumerableSource { + + @Serial + private static final long serialVersionUID = 773832461044750722L; + + final DeferredEnumerableSource enumerable; + + SkipStreamerDeferredEnumerable(Streamer upstream, DeferredEnumerableSource enumerable, long count) { + super(upstream, count); + this.enumerable = enumerable; + } + + @Override + public boolean nextSync() throws Throwable { + while (remaining-- > 0) { + if (!enumerable.nextSync()) { + return false; + } + } + return enumerable.nextSync(); + } + + @Override + public CompletionStage enumerableReady() { + return enumerable.enumerableReady(); + } + + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTake.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTake.java index 7f0a20371b..74563c2479 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTake.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTake.java @@ -17,35 +17,42 @@ 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.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.operators.*; public record StreamableTake(Streamable source, long count) implements Streamable, HasUpstreamStreamableSource { + @SuppressWarnings("unchecked") @Override public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { - var dsc = cancellation.derive(); - return new TakeStreamer<>(source.stream(dsc), count, dsc); + var upstream = source.stream(cancellation); + if (upstream instanceof IndexableSource isrc) { + return new TakeStreamerIndexable<>(upstream, (IndexableSource)isrc, count); + } else + if (upstream instanceof DeferredEnumerableSource dsrc) { + return new TakeStreamerDeferredEnumerable<>(upstream, (DeferredEnumerableSource)dsrc, count); + } else + if (upstream instanceof EnumerableSource esrc) { + return new TakeStreamerEnumerable<>(upstream, (EnumerableSource)esrc, count); + } + return new TakeStreamerBasic<>(upstream, count); } - static final class TakeStreamer implements Streamer { + static abstract class TakeStreamerBase implements Streamer { final Streamer upstream; - final Disposable upstreamDisposable; - long remaining; - TakeStreamer(Streamer upstream, long count, Disposable upstreamDisposable) { + TakeStreamerBase(Streamer upstream, long count) { this.upstream = upstream; - this.upstreamDisposable = upstreamDisposable; this.remaining = count; } @Override public @NonNull CompletionStage next() { if (remaining-- <= 0L) { - upstreamDisposable.dispose(); return NEXT_FALSE; } return upstream.next(); @@ -61,4 +68,87 @@ static final class TakeStreamer implements Streamer { return upstream.finish(); } } + + static final class TakeStreamerBasic extends TakeStreamerBase { + + TakeStreamerBasic(Streamer upstream, long count) { + super(upstream, count); + } + + } + + static final class TakeStreamerIndexable extends TakeStreamerBase + implements IndexableSource { + + final IndexableSource indexable; + + final long count; + + TakeStreamerIndexable(Streamer upstream, IndexableSource indexable, long count) { + super(upstream, count); + this.indexable = indexable; + this.count = count; + } + + @Override + public @NonNull T elementAt(long index) throws Throwable { + return indexable.elementAt(index); + } + + @Override + public long limit() { + return Math.min(count, indexable.limit()); + } + } + + static final class TakeStreamerEnumerable extends TakeStreamerBase + implements EnumerableSource { + + final EnumerableSource enumerable; + + final long count; + + TakeStreamerEnumerable(Streamer upstream, EnumerableSource enumerable, long count) { + super(upstream, count); + this.enumerable = enumerable; + this.count = count; + } + + @Override + public boolean nextSync() throws Throwable { + while (remaining-- > 0) { + return enumerable.nextSync(); + } + return false; + } + + } + + static final class TakeStreamerDeferredEnumerable extends TakeStreamerBase + implements DeferredEnumerableSource { + + final DeferredEnumerableSource enumerable; + + final long count; + + TakeStreamerDeferredEnumerable(Streamer upstream, DeferredEnumerableSource enumerable, long count) { + super(upstream, count); + this.enumerable = enumerable; + this.count = count; + } + + @Override + public boolean nextSync() throws Throwable { + while (remaining-- > 0) { + return enumerable.nextSync(); + } + return false; + } + + @Override + public CompletionStage enumerableReady() { + return enumerable.enumerableReady(); + } + + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DisposableStreamerCancellationTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DisposableStreamerCancellationTest.java new file mode 100644 index 0000000000..4296f44282 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DisposableStreamerCancellationTest.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.internal.operators.streamable; + +import static org.junit.jupiter.api.Assertions.*; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.internal.disposables.NeverDisposableStreamerCancellation; +import io.reactivex.rxjava4.testsupport.TestHelper; + +public class DisposableStreamerCancellationTest extends StreamableBaseTest { + + @Test + public void never() throws Throwable { + TestHelper.checkEnum(NeverDisposableStreamerCancellation.class); + + var ndsc = DisposableStreamerCancellation.never(); + + assertTrue(ndsc.add(Disposable.empty())); + assertTrue(ndsc.remove(Disposable.empty())); + assertTrue(ndsc.delete(Disposable.empty())); + assertFalse(ndsc.isDisposed()); + ndsc.dispose(); + assertFalse(ndsc.isDisposed()); + assertSame(ndsc, ndsc.derive()); + } + +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java index 00980ace8d..63349a08b2 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java @@ -22,6 +22,7 @@ import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; import io.reactivex.rxjava4.exceptions.TestException; public class StreamableBlockingLastTest extends StreamableBaseTest { @@ -75,4 +76,9 @@ public void nextAndFinishCrash() throws Throwable { assertTrue(ex.getSuppressed()[0] instanceof TestException, "Wrong exception? " + ex.getSuppressed()[0]); } + @Test + public void normalCancellation() throws Throwable { + assertEquals(1, Streamable.just(1).blockingLast(DisposableStreamerCancellation.never())); + } + } 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 5b01348178..cfbb1285af 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 @@ -18,7 +18,6 @@ import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.Streamable; -import io.reactivex.rxjava4.exceptions.TestException; public class StreamableFromStreamTest extends StreamableBaseTest { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java index c810d4d51d..58650a2574 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java @@ -13,12 +13,16 @@ 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.processors.DispatchStreamProcessor; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableSkipTest extends StreamableBaseTest { @@ -85,4 +89,146 @@ public void skipMore() throws Throwable { .awaitDone(5, TimeUnit.SECONDS) .assertResult(); } + + @Test + public void normal2() throws Throwable { + Streamable.range(1, 5) + .hide() + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(3, 4, 5); + } + + @Test + public void normalIndexed() throws Throwable { + Streamable.range(1, 5) + .skip(2) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(3, 4, 5)); + } + + @Test + public void normalEnumerable() throws Throwable { + Streamable.range(1, 5) + .filter(v -> v >= 1) + .skip(2) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(3, 4, 5)); + } + + @Test + public void normalEnumerable2() throws Throwable { + Streamable.range(1, 5) + .filter(v -> v >= 4) + .skip(3) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of()); + } + + @Test + public void normalDeferredEnumerable() throws Throwable { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, v + 1, v + 2, v + 3, v + 4)) + .skip(2) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(3, 4, 5)); + } + + @Test + public void normalDeferredEnumerable2() throws Throwable { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, v + 1, v + 2, v + 3, v + 4)) + .skip(6) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of()); + } + + @Test + public void delayed() throws Throwable { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(3, 4, 5); + } + + @Test + public void intervalRange() throws Throwable { + Streamable.intervalRange(1, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(3L, 4L, 5L); + } + + @Test + public void dispatch() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp.skip(2).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.next(4).toCompletableFuture().join(); + dsp.finish(null).toCompletableFuture().join(); + + ts + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(3, 4); + } + + @Test + public void dispatchError() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp.skip(2).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.next(4).toCompletableFuture().join(); + dsp.finish(new TestException()).toCompletableFuture().join(); + + ts + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 3, 4); + } + + @Test + public void asyncCompletion() throws Throwable { + Streamable.fromCompletable(Completable.complete().delay(1, TimeUnit.MILLISECONDS)) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void asyncError() throws Throwable { + Streamable.fromCompletable(Completable.error(new TestException()).delay(1, TimeUnit.MILLISECONDS)) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeTest.java index f3b6b311ad..0d89beeef0 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeTest.java @@ -15,8 +15,10 @@ import static org.junit.jupiter.api.Assertions.*; +import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; @@ -96,4 +98,97 @@ public void cancelled() throws Throwable { .awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3); } + + @Test + public void normalHidden() throws Throwable { + Streamable.range(1, 10) + .hide() + .take(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void normalIndexed() throws Throwable { + Streamable.range(1, 10) + .take(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void normalEnumerable() throws Throwable { + Streamable.range(1, 10) + .filter(v -> v >= 2) + .take(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(2, 3, 4, 5, 6); + } + + @Test + public void normalEnumerableDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.range(1, 10) + .filter(v -> v >= 2) + .take(5) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(2, 3, 4, 5, 6); + }); + } + + @Test + public void normalDeferredEnumerable() throws Throwable { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, v + 1, v + 2, v + 3, v + 4, v + 5, v + 6)) + .take(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void normalIndexedCollect() throws Throwable { + Streamable.range(1, 10) + .take(5) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } + + @Test + public void normalIndexedCollect2() throws Throwable { + Streamable.range(1, 3) + .take(5) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3)); + } + + @Test + public void normalEnumerableCollect() throws Throwable { + Streamable.range(1, 10) + .filter(v -> v >= 2) + .take(5) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(2, 3, 4, 5, 6)); + } + + @Test + public void normalDeferredEnumerableCollect() throws Throwable { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, v + 1, v + 2, v + 3, v + 4, v + 5, v + 6)) + .take(5) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of(1, 2, 3, 4, 5)); + } } diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index bbc0c0b575..1a5dd34071 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -728,6 +728,8 @@ public void checkStreamable() { defaultValues.put(StreamSink.class, new DispatchStreamProcessor<>()); + defaultValues.put(StreamerCancellation.class, new CompositeDisposable()); + // TODO insert new config record types here @SuppressWarnings("rawtypes")