From e2fd472dc44c52969812c12f460698870b4f1227 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Thu, 9 Jul 2026 07:50:11 +0200 Subject: [PATCH 1/2] 4.x: Streamable ~ take native implementation --- .../io/reactivex/rxjava4/core/Streamable.java | 103 ++++++++++++++---- .../operators/streamable/StreamableTake.java | 66 +++++++++++ .../streamable/StreamableTakeTest.java | 48 +++++++- .../CheckParamValidationNamingTest.java | 1 + .../validators/CheckParamValidationTest.java | 2 + 5 files changed, 195 insertions(+), 25 deletions(-) create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTake.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 924342b0fc..77825167ab 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -13,17 +13,73 @@ package io.reactivex.rxjava4.core; -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicLong; -import java.util.stream.*; - -import io.reactivex.rxjava4.annotations.*; -import io.reactivex.rxjava4.core.config.*; -import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.functions.*; -import io.reactivex.rxjava4.internal.functions.ObjectHelper; -import io.reactivex.rxjava4.internal.operators.streamable.*; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Flow; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collector; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import io.reactivex.rxjava4.annotations.CheckReturnValue; +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.annotations.Nullable; +import io.reactivex.rxjava4.core.config.StandardConcurrentConfig; +import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; +import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.functions.BiConsumer; +import io.reactivex.rxjava4.functions.Consumer; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.functions.Predicate; +import io.reactivex.rxjava4.functions.Supplier; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableCollector; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableConcatIterable; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableDefer; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableDelay; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableEmpty; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableError; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFilter; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFlatMap; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableForEach; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromArray; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromCompletable; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromIterable; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromMaybe; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromPublisher; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromSingle; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromStream; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableGroupBy; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableHide; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableIntercept; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableIntervalRange; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableJust; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableLift; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableMap; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableMapOptional; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableNever; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableOnErrorResumeNext; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableRange; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableRangeLong; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableTake; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableTakeUntil; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableTakeWhile; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableTimeout; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableTimer; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableToObservable; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableUsing; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableZip; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.subscribers.TestSubscriber; @@ -806,23 +862,22 @@ default Streamable onErrorResumeNext(@NonNull Function + * Note that cancellation of the upstream happens when the downstream + * calls {@link Streamer#next()} because unlike the push-based {@code take} + * implementations, the current upstream value has to remain accessible until + * the downstream calls {@code next} or {@link Streamer#finish()}. + * @param count the maximum number of items to relay * @return the new {@code Streamable} instance - * @throws IllegalArgumentException if {@code n} is non-positive + * @throws IllegalArgumentException if {@code count} is negative */ @CheckReturnValue @NonNull - default Streamable take(long n) { - ObjectHelper.verifyPositive(n, "n"); - return defer(() -> { - var countdown = new AtomicLong(n); - return transform((item, emitter, stopper) -> { - emitter.emit(item); - if (countdown.decrementAndGet() <= 0) { - stopper.dispose(); - } - }); - }); + default Streamable take(long count) { + if (count < 0) { + throw new IllegalArgumentException("count >= 0 required but it was " + count); + } + return RxJavaPlugins.onAssembly(new StreamableTake<>(this, count)); } /** 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 new file mode 100644 index 0000000000..545fd9bdc5 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTake.java @@ -0,0 +1,66 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.concurrent.CompletionStage; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.Streamer; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; + +public record StreamableTake(Streamable source, long count) +implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { + var dsc = cancellation.derive(); + return new TakeStreamer<>(source.stream(dsc), count, dsc); + } + + static final class TakeStreamer implements Streamer { + final Streamer upstream; + + final Disposable upstreamDisposable; + + long remaining; + + TakeStreamer(Streamer upstream, long count, Disposable upstreamDisposable) { + 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(); + } + + @Override + public @NonNull T current() { + return upstream.current(); + } + + @Override + public @NonNull CompletionStage finish() { + return upstream.finish(); + } + } +} 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 1dc3a8a001..e2c3bccd70 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 @@ -13,13 +13,18 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; + import io.reactivex.rxjava4.core.Flowable; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; public class StreamableTakeTest extends StreamableBaseTest { @@ -52,4 +57,45 @@ public void fewer() throws Throwable { assertFalse(isCancelled.get(), "Cancel was propagated!"); } + + @Test + public void error() { + Streamable.error(new TestException()) + .take(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void doubleTake() { + Streamable.range(1, 5) + .take(3) + .take(1) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1) + ; + } + + @Test + public void cancelled() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp.take(3).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(); + + awaitNoStreamers(dsp, 1000); + + ts + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3); + } } diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java index 4bcdabed08..8a67892d70 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java @@ -318,6 +318,7 @@ static void processFile(Class clazz) throws Exception { if (linek.startsWith("public") || linek.startsWith("private") || linek.startsWith("protected") || linek.startsWith("static") + || linek.startsWith("default") || linek.startsWith(baseClassName)) { break; } diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index 89541434e0..9b72fd1501 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -634,6 +634,8 @@ public void checkStreamable() { addIgnore(new ParamIgnore(Streamable.class, "rangeLong", Long.TYPE, Long.TYPE)); addIgnore(new ParamIgnore(Streamable.class, "intervalRange", Long.TYPE, Long.TYPE, Long.TYPE, Long.TYPE, TimeUnit.class, Scheduler.class)); addIgnore(new ParamIgnore(Streamable.class, "intervalRange", Long.TYPE, Long.TYPE, Long.TYPE, Long.TYPE, TimeUnit.class, ExecutorService.class)); + // zero take is allowed + addOverride(new ParamOverride(Streamable.class, 0, ParamMode.NON_NEGATIVE, "take", Long.TYPE)); // ----------------------------------------------------------------------------------- From b7026f279d4758c3a4c37e03549c08a50d628b54 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Thu, 9 Jul 2026 07:54:52 +0200 Subject: [PATCH 2/2] fix import wildcard, forgot to config eclipse --- .../io/reactivex/rxjava4/core/Streamable.java | 76 +++---------------- .../operators/streamable/StreamableTake.java | 6 +- .../streamable/StreamableTakeTest.java | 6 +- 3 files changed, 13 insertions(+), 75 deletions(-) diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 77825167ab..706b0b94d8 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -13,73 +13,15 @@ package io.reactivex.rxjava4.core; -import java.util.List; -import java.util.Objects; -import java.util.Optional; -import java.util.concurrent.Callable; -import java.util.concurrent.CancellationException; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Flow; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collector; -import java.util.stream.Collectors; -import java.util.stream.Stream; - -import io.reactivex.rxjava4.annotations.CheckReturnValue; -import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.annotations.Nullable; -import io.reactivex.rxjava4.core.config.StandardConcurrentConfig; -import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; -import io.reactivex.rxjava4.disposables.CompositeDisposable; -import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; -import io.reactivex.rxjava4.disposables.StreamerCancellation; -import io.reactivex.rxjava4.functions.BiConsumer; -import io.reactivex.rxjava4.functions.Consumer; -import io.reactivex.rxjava4.functions.Function; -import io.reactivex.rxjava4.functions.Predicate; -import io.reactivex.rxjava4.functions.Supplier; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableCollector; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableConcatIterable; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableDefer; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableDelay; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableEmpty; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableError; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFilter; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFlatMap; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableForEach; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromArray; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromCompletable; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromIterable; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromMaybe; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromPublisher; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromSingle; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromStream; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableGroupBy; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableHide; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableIntercept; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableIntervalRange; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableJust; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableLift; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableMap; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableMapOptional; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableNever; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableOnErrorResumeNext; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableRange; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableRangeLong; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableTake; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableTakeUntil; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableTakeWhile; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableTimeout; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableTimer; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableToObservable; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableUsing; -import io.reactivex.rxjava4.internal.operators.streamable.StreamableZip; +import java.util.*; +import java.util.concurrent.*; +import java.util.stream.*; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.config.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.functions.*; +import io.reactivex.rxjava4.internal.operators.streamable.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.subscribers.TestSubscriber; 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 545fd9bdc5..7f0a20371b 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 @@ -16,10 +16,8 @@ import java.util.concurrent.CompletionStage; import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.core.Streamable; -import io.reactivex.rxjava4.core.Streamer; -import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; public record StreamableTake(Streamable source, long count) 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 e2c3bccd70..f3b6b311ad 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 @@ -13,16 +13,14 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; -import io.reactivex.rxjava4.core.Flowable; -import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.processors.DispatchStreamProcessor;