Skip to content

4.x: Streamable +collect +blocking +withCancel #8333

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
akarnokd merged 1 commit into from
Oct 15, 2026
+879 −42
Merged

4.x: Streamable +collect +blocking +withCancel #8333

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 142 additions & 6 deletions src/main/java/io/reactivex/rxjava4/core/Streamable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,21 @@ public interface Streamable<@NonNull T> {
return RxJavaPlugins.onAssembly(new StreamableError<>(throwable));
}

/**
* Creates a {@code Streamable} that invokes the given {@link Supplier} to produce a {@link Throwable}
* that will be immediately signaled when the streaming begins.
* @param <T> the element type of the sequence
* @param throwableSupplier the {@code Supplier} to produce a {@code Throwable} for each of the streamers.
* @return the new {@code Streamable} instance
* @throws NullPointerException if {@code throwableSupplier} is {@code null}
*/
@CheckReturnValue
@NonNull
static <@NonNull T> Streamable<T> error(Supplier<? extends Throwable> throwableSupplier) {
Objects.requireNonNull(throwableSupplier, "throwableSupplier is null");
return defer(() -> error(throwableSupplier.get()));
}

/**
* Returns an empty {@code Streamable} that never produces an item and just completes.
* @param <T> the element type
Expand Down Expand Up @@ -629,8 +644,25 @@ static <T> Streamable<List<T>> zip(Iterable<? extends Streamable<? extends T>> s
@CheckReturnValue
@NonNull
default T blockingFirst() {
return StreamableBlocking.blockingFirst(this);
return StreamableBlocking.blockingFirst(this, null, new CompositeDisposable());
}

/**
* Blocks the current thread until this {@code Streamable} produces one item, which is then returned immediately,
* or the {@code defaultItem} is returned if the upstream is empty.
* @param defaultItem if the upstream is empty
* @return the first item of this {@code Streamable}
* @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 defaultItem} is {@code null}
*/
@CheckReturnValue
@NonNull
default T blockingFirst(@NonNull T defaultItem) {
Objects.requireNonNull(defaultItem, "defaultItem is null");
return StreamableBlocking.blockingFirst(this, defaultItem, new CompositeDisposable());
}

/**
* 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
Expand All @@ -642,9 +674,27 @@ default T blockingFirst() {
*/
@CheckReturnValue
@NonNull
default T blockingFirst(StreamerCancellation cancellation) {
default T blockingFirst(@NonNull StreamerCancellation cancellation) {
Objects.requireNonNull(cancellation, "cancellation is null");
return StreamableBlocking.blockingFirst(this, null, cancellation);
}

/**
* Blocks the current thread until this {@code Streamable} produces one item, which is then returned immediately,
* or the {@code defaultItem} is returned if the upstream is empty.
* @param defaultItem if the upstream is empty
* @param cancellation the external cancellation resource to pass into the chain
* @return the first item of this {@code Streamable}
* @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 defaultItem} or {@code cancellation} is {@code null}
*/
@CheckReturnValue
@NonNull
default T blockingFirst(@NonNull T defaultItem, @NonNull StreamerCancellation cancellation) {
Objects.requireNonNull(defaultItem, "defaultItem is null");
Objects.requireNonNull(cancellation, "cancellation is null");
return StreamableBlocking.blockingFirst(this, cancellation);
return StreamableBlocking.blockingFirst(this, defaultItem, cancellation);
}

/**
Expand All @@ -658,7 +708,25 @@ default T blockingFirst(StreamerCancellation cancellation) {
@CheckReturnValue
@NonNull
default T blockingLast() {
return StreamableBlocking.blockingLast(this);
return StreamableBlocking.blockingLast(this, null, new CompositeDisposable());
}

/**
* Blocks the current thread until this {@code Streamable} produces all of its items
* and the very last is then returned, or returns the {@code defaultItem} if the
* upstream is empty.
* @param defaultItem if the upstream is empty
* @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 defaultItem} is {@code null}
*/
@CheckReturnValue
@NonNull
default T blockingLast(@NonNull T defaultItem) {
Objects.requireNonNull(defaultItem, "defaultItem is null");
return StreamableBlocking.blockingLast(this, defaultItem, new CompositeDisposable());
}

/**
Expand All @@ -673,9 +741,29 @@ default T blockingLast() {
*/
@CheckReturnValue
@NonNull
default T blockingLast(StreamerCancellation cancellation) {
default T blockingLast(@NonNull StreamerCancellation cancellation) {
Objects.requireNonNull(cancellation, "cancellation is null");
return StreamableBlocking.blockingLast(this, cancellation);
return StreamableBlocking.blockingLast(this, null, cancellation);
}

/**
* Blocks the current thread until this {@code Streamable} produces all of its items
* and the very last is then returned, or returns the {@code defaultItem} if the
* upstream is empty.
* @param defaultItem if the upstream is empty
* @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 defaultItem} or {@code cancellation} is {@code null}
*/
@CheckReturnValue
@NonNull
default T blockingLast(@NonNull T defaultItem, @NonNull StreamerCancellation cancellation) {
Objects.requireNonNull(defaultItem, "defaultItem is null");
Objects.requireNonNull(cancellation, "cancellation is null");
return StreamableBlocking.blockingLast(this, defaultItem, cancellation);
}

/**
Expand All @@ -696,6 +784,24 @@ default <A, R> Streamable<R> collect(@NonNull Collector<T, A, R> collector) {
return RxJavaPlugins.onAssembly(new StreamableCollector<>(this, collector));
}

/**
* Collects all upstream values into a data structure, provided for each streamer, via
* a custom collector callback and emits this single item as the result of the returned
* {@code Streamable}.
* @param <U> the element type of the data structure and returned value
* @param initialItemSupplier provides an individual item for each streamer to collect upstream items into
* @param collector the callback that receives the individual data structure item and the current upstream value
* @return the new {@code Streamable} instance
* @throws NullPointerException if {@code initialItemSupplier} or {@code collector} is {@code null}
*/
@CheckReturnValue
@NonNull
default <U> Streamable<U> collect(Supplier<? extends U> initialItemSupplier, BiConsumer<? super U, ? super T> collector) {
Objects.requireNonNull(initialItemSupplier, "initialItemSupplier is null");
Objects.requireNonNull(collector, "collector is null");
return RxJavaPlugins.onAssembly(new StreamableCollect<>(this, initialItemSupplier, collector));
}

/**
* Delays the delivery of each upstream item by the given time amount.
* @param time the delay time
Expand Down Expand Up @@ -1191,6 +1297,36 @@ default Observable<T> toObservable() {
.await(), executor);
}

/**
* Replaces the {@link StreamerCancellation} instance of the downstream travelling through
* this operator with the provided instance to allow external cancellation.
* @param cancellation the new {@code StreamerCancellation} to use
* @return the new {@code Streamable} instance
* @throws NullPointerException if {@code cancellation} is {@code null}
*/
@CheckReturnValue
@NonNull
default Streamable<T> withCancellation(StreamerCancellation cancellation) {
Objects.requireNonNull(cancellation, "cancellation is null");
return withCancellation(_ -> cancellation);
}

/**
* Replaces the {@link StreamerCancellation} instance of the downstream travelling through
* this operator with the instance returned by the {@code onStream} callback to allow external,
* possibly dependent cancellation.
* @param onStream the function called with the downstream {@code StreamerCancellation} and expected
* to return a {@code StreamerCancellation} to pass it to the upstream source
* @return the new {@code Streamable} instance
* @throws NullPointerException if {@code onStream} is {@code null}
*/
@CheckReturnValue
@NonNull
default Streamable<T> withCancellation(Function<? super StreamerCancellation, ? extends StreamerCancellation> onStream) {
Objects.requireNonNull(onStream, "onStream is null");
return RxJavaPlugins.onAssembly(new StreamableWithCancellation<>(this, onStream));
}

// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
// Consumption methods and outgoing converters
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
Expand Down
45 changes: 13 additions & 32 deletions src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,32 +26,19 @@ public record StreamableBlocking() {

/**
* Consumes the first item and finishes the {@link Streamable},
* throwing {@link NoSuchElementException} if the source is empty.
* @param <T> the element type
* @param source the source {@code Streamable}
* @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> T blockingFirst(Streamable<T> source) {
return blockingFirst(source, new CompositeDisposable());
}

/**
* Consumes the first item and finishes the {@link Streamable},
* throwing {@link NoSuchElementException} if the source is empty.
* throwing {@link NoSuchElementException} if the source is empty and {@code defaultItem} is {@code null}.
* @param <T> the element type
* @param source the source {@code Streamable}
* @param defaultItem if not null and the source is empty, this value will be returned
* @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> T blockingFirst(Streamable<T> source, StreamerCancellation cancellation) {
public static <T> T blockingFirst(@NonNull Streamable<T> source,
@Nullable T defaultItem, @NonNull StreamerCancellation cancellation) {
var streamer = source.stream(cancellation);
Throwable nextException = null;
Throwable finishException = null;
Expand All @@ -75,35 +62,26 @@ public static <T> T blockingFirst(Streamable<T> source, StreamerCancellation can
throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrapAndCombine(nextException, finishException));
}
if (result == null) {
if (defaultItem != null) {
return defaultItem;
}
throw new NoSuchElementException();
}
return result;
}

/**
* Consumes all upstream items and returns the very last or throws
* a {@link NoSuchElementException}.
* @param <T> the element type
* @param source the source sequence
* @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> T blockingLast(Streamable<T> source) {
return blockingLast(source, new CompositeDisposable());
}

/**
* Consumes all upstream items and returns the very last or throws
* a {@link NoSuchElementException}.
* a {@link NoSuchElementException} if {@code defaultItem} is also {@code null}.
* @param <T> the element type
* @param source the source sequence
* @param defaultItem if not null, item to return if the source is empty
* @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> T blockingLast(Streamable<T> source, StreamerCancellation cancellation) {
public static <T> T blockingLast(Streamable<T> source, T defaultItem, StreamerCancellation cancellation) {
var streamer = source.stream(cancellation);
Throwable nextException = null;
Throwable finishException = null;
Expand All @@ -127,6 +105,9 @@ public static <T> T blockingLast(Streamable<T> source, StreamerCancellation canc
throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrapAndCombine(nextException, finishException));
}
if (result == null) {
if (defaultItem != null) {
return defaultItem;
}
throw new NoSuchElementException();
}
return result;
Expand Down
Loading
Loading