Skip to content

Commit 93d6ddd

Browse files
akarnokdakarnokd
authored
4.x: Streamable + range, fromArray, take, error, defer (#8206)
* 4.x: Streamable + range, fromArray, take, error, defer * improve test coverage * Improve coverage * More organization, reclaim some coverage too * More tests and API clarifications * Style fixes * impossibly flaky unit test? overlapBackpressure2
1 parent e8ec405 commit 93d6ddd

85 files changed

Lines changed: 1546 additions & 437 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎src/main/java/io/reactivex/rxjava4/core/CompletionStageDisposable.java‎

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,14 @@
1616
import java.io.Serial;
1717
import java.lang.ref.Cleaner;
1818
import java.util.Objects;
19-
import java.util.concurrent.CompletionStage;
19+
import java.util.concurrent.*;
2020
import java.util.concurrent.atomic.AtomicBoolean;
2121
import java.util.function.Consumer;
2222

2323
import io.reactivex.rxjava4.annotations.NonNull;
2424
import io.reactivex.rxjava4.disposables.*;
25-
import io.reactivex.rxjava4.internal.util.AwaitCoordinatorStatic;
25+
import io.reactivex.rxjava4.exceptions.ThrowableWrapper;
26+
import io.reactivex.rxjava4.internal.util.*;
2627
import io.reactivex.rxjava4.plugins.RxJavaPlugins;
2728

2829
/**
@@ -84,19 +85,30 @@ public CompletionStageDisposable(@NonNull CompletionStage<T> stage, @NonNull Dis
8485
}
8586
/**
8687
* Await the completion of the current stage.
88+
* <p>
89+
* Rethrows any original unchecked exceptions as is.
90+
* @throws CancellationException if the computation was cancelled
91+
* @throws ThrowableWrapper if the original exception was a checked exception
8792
*/
8893
public void await() {
89-
state.lazySet(true);
90-
AwaitCoordinatorStatic.await(stage);
94+
await(null);
9195
}
9296

9397
/**
9498
* Await the completion of the current stage.
99+
* <p>
100+
* Rethrows any original unchecked exceptions as is.
95101
* @param canceller the canceller link
102+
* @throws CancellationException if the computation was cancelled
103+
* @throws ThrowableWrapper if the original exception was a checked exception
96104
*/
97105
public void await(DisposableContainer canceller) {
98106
state.lazySet(true);
99-
AwaitCoordinatorStatic.await(stage, canceller);
107+
try {
108+
AwaitCoordinatorStatic.await(stage, canceller);
109+
} catch (CompletionException ce) {
110+
throw ExceptionHelper.wrapOrThrow(ce.getCause());
111+
}
100112
}
101113

102114
/**

‎src/main/java/io/reactivex/rxjava4/core/Flowable.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18059,7 +18059,7 @@ public final Stream<T> blockingStream(int prefetch) {
1805918059
Objects.requireNonNull(transformer, "transformer is null");
1806018060
Objects.requireNonNull(scheduler, "scheduler is null");
1806118061
ObjectHelper.verifyPositive(prefetch, "prefetch");
18062-
return new FlowableVirtualTransformExecutor<>(this, transformer, null, scheduler, prefetch);
18062+
return RxJavaPlugins.onAssembly(new FlowableVirtualTransformExecutor<>(this, transformer, null, scheduler, prefetch));
1806318063
}
1806418064

1806518065
/**
@@ -18097,7 +18097,7 @@ public final Stream<T> blockingStream(int prefetch) {
1809718097
Objects.requireNonNull(transformer, "transformer is null");
1809818098
Objects.requireNonNull(executor, "executor is null");
1809918099
ObjectHelper.verifyPositive(prefetch, "prefetch");
18100-
return new FlowableVirtualTransformExecutor<>(this, transformer, executor, null, prefetch);
18100+
return RxJavaPlugins.onAssembly(new FlowableVirtualTransformExecutor<>(this, transformer, executor, null, prefetch));
1810118101
}
1810218102

1810318103
/**
@@ -18148,6 +18148,6 @@ public final Streamable<T> toStreamable() {
1814818148
@NonNull
1814918149
public final Streamable<T> toStreamable(ExecutorService executor) {
1815018150
Objects.requireNonNull(executor, "executor is null");
18151-
return new StreamableFromPublisher<>(this, executor);
18151+
return RxJavaPlugins.onAssembly(new StreamableFromPublisher<>(this, executor));
1815218152
}
1815318153
}

‎src/main/java/io/reactivex/rxjava4/core/Streamable.java‎

Lines changed: 128 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -13,16 +13,18 @@
1313

1414
package io.reactivex.rxjava4.core;
1515

16-
import java.lang.reflect.InvocationTargetException;
1716
import java.util.*;
1817
import java.util.concurrent.*;
18+
import java.util.concurrent.atomic.AtomicLong;
1919

2020
import io.reactivex.rxjava4.annotations.*;
2121
import io.reactivex.rxjava4.disposables.*;
2222
import io.reactivex.rxjava4.exceptions.Exceptions;
2323
import io.reactivex.rxjava4.functions.*;
24+
import io.reactivex.rxjava4.internal.functions.ObjectHelper;
2425
import io.reactivex.rxjava4.internal.operators.streamable.*;
25-
import io.reactivex.rxjava4.internal.util.AwaitCoordinatorStatic;
26+
import io.reactivex.rxjava4.internal.util.*;
27+
import io.reactivex.rxjava4.plugins.RxJavaPlugins;
2628
import io.reactivex.rxjava4.schedulers.Schedulers;
2729
import io.reactivex.rxjava4.subscribers.TestSubscriber;
2830

@@ -75,7 +77,7 @@ default Streamer<T> stream() {
7577
@CheckReturnValue
7678
@NonNull
7779
static <@NonNull T> Streamable<T> empty() {
78-
return new StreamableEmpty<>();
80+
return RxJavaPlugins.onAssembly(new StreamableEmpty<>());
7981
}
8082

8183
/**
@@ -88,7 +90,21 @@ default Streamer<T> stream() {
8890
@NonNull
8991
static <@NonNull T> Streamable<T> just(@NonNull T item) {
9092
Objects.requireNonNull(item, "item is null");
91-
return new StreamableJust<>(item);
93+
return RxJavaPlugins.onAssembly(new StreamableJust<>(item));
94+
}
95+
96+
/**
97+
* Streams all elements of the given items array.
98+
* @param <T> the element type of the items
99+
* @param items the array of items to stream
100+
* @return the new {@code Streamable} instance
101+
* @throws NullPointerException if {@code items} is {@code null}
102+
*/
103+
@SafeVarargs
104+
@NonNull
105+
static <@NonNull T> Streamable<T> fromArray(T... items) {
106+
Objects.requireNonNull(items, "items is null");
107+
return RxJavaPlugins.onAssembly(new StreamableFromArray<>(items));
92108
}
93109

94110
/**
@@ -115,7 +131,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source) {
115131
@NonNull
116132
static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNull ExecutorService executor) {
117133
Objects.requireNonNull(source, "source is null");
118-
return new StreamableFromPublisher<>(source, executor);
134+
return RxJavaPlugins.onAssembly(new StreamableFromPublisher<>(source, executor));
119135
}
120136

121137
/**
@@ -193,6 +209,32 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
193209
}, executor);
194210
}
195211

212+
/**
213+
* Defers the creation of the actual {@code Streamable}, allowing a per streamer
214+
* state to be created along with it.
215+
* @param <T> the element type of the {@code Streamable}
216+
* @param supplier the callback that returns the actual {@code Streamable} to be streamed
217+
* @return the new {@code Streamable} instance
218+
* @throws NullPointerException if {@code supplier} is {@code null}
219+
*/
220+
static <@NonNull T> Streamable<T> defer(Supplier<? extends Streamable<? extends T>> supplier) {
221+
Objects.requireNonNull(supplier, "supplier is null");
222+
return RxJavaPlugins.onAssembly(new StreamableDefer<>(supplier));
223+
}
224+
225+
/**
226+
* Creates a {@code Streamable} that signals the given {@link Throwable} immediately when it begins streaming,
227+
* ending the sequence.
228+
* @param <T> the element type of the sequence
229+
* @param throwable the {@code Throwable} to signal immediately upon streaming
230+
* @return the new {@code Streamable} instance
231+
* @throws NullPointerException if {@code throwable} is {@code null}
232+
*/
233+
static <@NonNull T> Streamable<T> error(Throwable throwable) {
234+
Objects.requireNonNull(throwable, "throwable is null");
235+
return RxJavaPlugins.onAssembly(new StreamableError<>(throwable));
236+
}
237+
196238
/**
197239
* Emits the elements of each inner sequence produced by the outher sequence.
198240
* @param <T> the common element type
@@ -212,6 +254,54 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
212254
}, exec);
213255
}
214256

257+
/**
258+
* Emits elements from start up to start + count exclusive.
259+
* @param start the start element
260+
* @param count the number of elements to emit
261+
* @return the new {@code Streamable} instance
262+
*/
263+
static Streamable<Integer> range(int start, int count) {
264+
if (count < 0) {
265+
throw new IllegalArgumentException("count >= 0 required but it was " + count);
266+
} else
267+
if (count == 0) {
268+
return empty();
269+
} else
270+
if (count == 1) {
271+
return just(start);
272+
} else
273+
if ((long)start + (count - 1) > Integer.MAX_VALUE) {
274+
throw new IllegalArgumentException("Integer overflow");
275+
}
276+
return RxJavaPlugins.onAssembly(new StreamableRange(start, count));
277+
}
278+
279+
/**
280+
* Emits elements from start up to start + count exclusive.
281+
* @param start the start element
282+
* @param count the number of elements to emit
283+
* @return the new {@code Streamable} instance
284+
*/
285+
static Streamable<Long> rangeLong(long start, long count) {
286+
if (count < 0) {
287+
throw new IllegalArgumentException("count >= 0 required but it was " + count);
288+
}
289+
290+
if (count == 0) {
291+
return empty();
292+
}
293+
294+
if (count == 1) {
295+
return just(start);
296+
}
297+
298+
long end = start + (count - 1);
299+
if (start > 0 && end < 0) {
300+
throw new IllegalArgumentException("Overflow! start + count is bigger than Long.MAX_VALUE");
301+
}
302+
return RxJavaPlugins.onAssembly(new StreamableRangeLong(start, count));
303+
}
304+
215305
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
216306
// Operators
217307
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
@@ -274,6 +364,25 @@ default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
274364
.await(emitter.canceller()), executor);
275365
}
276366

367+
/**
368+
* Takes at most the given number of items from the upstream and relays it to the downstream,
369+
* then cancels the rest of the sequence.
370+
* @param n the maximum number of items to relay
371+
* @return the new {@code Streamable} instance
372+
*/
373+
default Streamable<T> take(long n) {
374+
ObjectHelper.verifyPositive(n, "n");
375+
return defer(() -> {
376+
var countdown = new AtomicLong(n);
377+
return transform((item, emitter, stopper) -> {
378+
emitter.emit(item);
379+
if (countdown.decrementAndGet() <= 0) {
380+
stopper.dispose();
381+
}
382+
});
383+
});
384+
}
385+
277386
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
278387
// Consumption methods and outgoing converters
279388
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
@@ -324,13 +433,12 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
324433
*/
325434
@CheckReturnValue
326435
@NonNull
327-
@SuppressWarnings("unchecked")
328436
default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer, @NonNull DisposableContainer canceller, @NonNull ExecutorService executor) {
329437
Objects.requireNonNull(consumer, "consumer is null");
330438
Objects.requireNonNull(canceller, "canceller is null");
331439
Objects.requireNonNull(executor, "executor is null");
332440
final Streamable<T> me = this;
333-
var future = executor.submit(() -> {
441+
var future = CompletableFuture.<Void>supplyAsync(() -> {
334442
try (var str = me.stream(canceller)) {
335443
while (!canceller.isDisposed()) {
336444
if (str.awaitNext(canceller)) {
@@ -344,20 +452,15 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
344452
// System.out.println("Canceller status after loop: " + canceller.isDisposed());
345453
} catch (final Throwable crash) {
346454
Exceptions.throwIfFatal(crash);
347-
// System.out.println("Canceller status in error: " + canceller.isDisposed());
348-
crash.printStackTrace();
349-
if (crash instanceof RuntimeException ex) {
350-
throw ex;
351-
}
352-
if (crash instanceof Exception ex) {
353-
throw ex;
455+
if (crash instanceof CompletionException ce) {
456+
throw ExceptionHelper.wrapOrThrow(ce.getCause());
354457
}
355-
throw new InvocationTargetException(crash);
458+
throw ExceptionHelper.wrapOrThrow(crash);
356459
}
357460
return null;
358-
});
461+
}, executor);
359462
canceller.add(Disposable.fromFuture(future));
360-
return new CompletionStageDisposable<>(StreamableHelper.toCompletionStage((Future<Void>) (Future<?>) future), canceller);
463+
return new CompletionStageDisposable<>(future, canceller);
361464
}
362465

363466
/**
@@ -369,7 +472,6 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
369472
*/
370473
@CheckReturnValue
371474
@NonNull
372-
@SuppressWarnings("unchecked")
373475
default CompletionStageDisposable<Void> forEach(
374476
@NonNull BiConsumer<? super T, ? super Disposable> consumer,
375477
@NonNull DisposableContainer canceller,
@@ -378,7 +480,7 @@ default CompletionStageDisposable<Void> forEach(
378480
Objects.requireNonNull(canceller, "canceller is null");
379481
Objects.requireNonNull(executor, "executor is null");
380482
final Streamable<T> me = this;
381-
var future = executor.submit(() -> {
483+
var future = CompletableFuture.<Void>supplyAsync(() -> {
382484
try (var str = me.stream(canceller)) {
383485
var stopper = Disposable.empty();
384486
while (!canceller.isDisposed() && !stopper.isDisposed()) {
@@ -394,21 +496,15 @@ default CompletionStageDisposable<Void> forEach(
394496
// System.out.println("Canceller status after loop: " + canceller.isDisposed());
395497
} catch (final Throwable crash) {
396498
Exceptions.throwIfFatal(crash);
397-
// System.out.println("Canceller status in error: " + canceller.isDisposed());
398-
// crash.printStackTrace();
399-
if (crash instanceof RuntimeException ex) {
400-
throw ex;
401-
}
402-
if (crash instanceof Exception ex) {
403-
throw ex;
499+
if (crash instanceof CompletionException ce) {
500+
throw ExceptionHelper.wrapOrThrow(ce.getCause());
404501
}
405-
throw new InvocationTargetException(crash);
502+
throw ExceptionHelper.wrapOrThrow(crash);
406503
}
407504
return null;
408505
});
409506
canceller.add(Disposable.fromFuture(future));
410-
return new CompletionStageDisposable<>(
411-
StreamableHelper.toCompletionStage((Future<Void>) (Future<?>) future), canceller);
507+
return new CompletionStageDisposable<>(future, canceller);
412508
}
413509

414510
/**
@@ -419,9 +515,8 @@ default CompletionStageDisposable<Void> forEach(
419515
default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull ExecutorService executor) {
420516
final Streamable<T> me = this;
421517
Flowable.<T>virtualCreate(emitter -> {
422-
// System.out.println("subscribe::virtualCreate");
423-
// System.out.println("subscribe::virtualCreate::forEach::emit");
424-
me.forEach(emitter::emit).await(emitter.canceller());
518+
me.forEach(emitter::emit, emitter.canceller().derive(), executor)
519+
.await(emitter.canceller());
425520
}, executor)
426521
.subscribe(subscriber);
427522
}
@@ -433,8 +528,7 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull
433528
default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
434529
final Streamable<T> me = this;
435530
Flowable.<T>virtualCreate(emitter -> {
436-
// System.out.println("Emitting " + v);
437-
me.forEach(emitter::emit).await(emitter.canceller());
531+
me.forEach(emitter::emit).await(emitter.canceller());
438532
})
439533
.subscribe(subscriber);
440534
}

‎src/main/java/io/reactivex/rxjava4/core/Streamer.java‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
package io.reactivex.rxjava4.core;
1515

1616
import java.util.*;
17-
import java.util.concurrent.CompletionStage;
17+
import java.util.concurrent.*;
1818

1919
import io.reactivex.rxjava4.annotations.NonNull;
2020
import io.reactivex.rxjava4.disposables.*;
@@ -197,4 +197,22 @@ default void awaitFinish() {
197197
default void awaitFinish(@NonNull DisposableContainer cancellation) {
198198
await(finish(cancellation), cancellation);
199199
}
200+
201+
/**
202+
* Use this constant in {@link #next(DisposableContainer)} to indicate
203+
* the next value is available, synchronously.
204+
*/
205+
CompletionStage<Boolean> NEXT_TRUE = CompletableFuture.completedStage(true);
206+
207+
/**
208+
* Use this constant in {@link #next(DisposableContainer)} to indicate
209+
* no more values will be available, synchronously.
210+
*/
211+
CompletionStage<Boolean> NEXT_FALSE = CompletableFuture.completedStage(false);
212+
213+
/**
214+
* Use this constant in {@link #finish(DisposableContainer)} to indicate
215+
* the cleanupp was done synchronously.
216+
*/
217+
CompletionStage<Void> FINISHED = CompletableFuture.completedStage(null);
200218
}

‎src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ private Exceptions() {
2828
}
2929
/**
3030
* Convenience method to throw a {@code RuntimeException} and {@code Error} directly
31-
* or wrap any other exception type into a {@code RuntimeException}.
31+
* or wrap any other exception type into a {@link ThrowableWrapper}.
3232
* @param t the exception to throw directly or wrapped
3333
* @return because {@code propagate} itself throws an exception or error, this is a sort of phantom return
3434
* value; {@code propagate} does not actually return anything

0 commit comments

Comments
 (0)