Skip to content

Commit 045b7ce

Browse files
akarnokdakarnokd
committed
4.x: Streamable map, filter, flatMap impl, fixes and reworks
1 parent f68a855 commit 045b7ce

14 files changed

Lines changed: 1098 additions & 106 deletions

File tree

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

Lines changed: 86 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import java.util.stream.Stream;
2020

2121
import io.reactivex.rxjava4.annotations.*;
22+
import io.reactivex.rxjava4.core.config.StandardConcurrentConfig;
2223
import io.reactivex.rxjava4.disposables.*;
2324
import io.reactivex.rxjava4.exceptions.Exceptions;
2425
import io.reactivex.rxjava4.functions.*;
@@ -84,6 +85,20 @@ public interface Streamable<@NonNull T> {
8485
return RxJavaPlugins.onAssembly(new StreamableJust<>(item));
8586
}
8687

88+
/**
89+
* Filters out the upstream items that do not pass the given predicate
90+
* @param predicate the callback that should return {@code true} to let the upstream value pass
91+
* or {@code false} to ignore it and continue with the next upstream item
92+
* @return the new {@code Streamable} instance
93+
* @throw NullPointerException if {@code predicate} is {@code null}
94+
*/
95+
@CheckReturnValue
96+
@NonNull
97+
default Streamable<T> filter(@NonNull Predicate<? super T> predicate) {
98+
Objects.requireNonNull(predicate, "predicate is null");
99+
return RxJavaPlugins.onAssembly(new StreamableFilter<>(this, predicate));
100+
}
101+
87102
/**
88103
* Streams all elements of the given items array.
89104
* @param <T> the element type of the items
@@ -427,6 +442,32 @@ default Streamable<T> hide() {
427442
return RxJavaPlugins.onAssembly(new StreamableHide<>(this));
428443
}
429444

445+
/**
446+
* Maps each upstream item into another item via a mapper function.
447+
* @param <R> the element type of the mapping
448+
* @param mapper the function that takes an upstream item and returns an item to be emitted
449+
* to the downstream
450+
* @return the new {@code Streamable} instance
451+
* @throw NullPointerException if {@code mapper} is {@code null}
452+
*/
453+
default <@NonNull R> Streamable<R> map(@NonNull Function<? super T, ? extends R> mapper) {
454+
Objects.requireNonNull(mapper, "mapper is null");
455+
return RxJavaPlugins.onAssembly(new StreamableMap<>(this, mapper));
456+
}
457+
458+
/**
459+
* Maps each upstream item into another, optional item via a mapper function that skips the empty optionals.
460+
* @param <R> the element type of the mapping
461+
* @param mapper the function that takes an upstream item and returns an optional item to be emitted / skipped
462+
* to the downstream
463+
* @return the new {@code Streamable} instance
464+
* @throw NullPointerException if {@code mapper} is {@code null}
465+
*/
466+
default <@NonNull R> Streamable<R> mapOptional(@NonNull Function<? super T, ? extends Optional<? extends R>> mapper) {
467+
Objects.requireNonNull(mapper, "mapper is null");
468+
return RxJavaPlugins.onAssembly(new StreamableMapOptional<>(this, mapper));
469+
}
470+
430471
/**
431472
* Transforms the upstream sequence into zero or more elements for the downstream.
432473
* @param <R> the result element type
@@ -469,6 +510,24 @@ default Streamable<T> take(long n) {
469510
});
470511
}
471512

513+
/**
514+
* Maps each upstream item onto a {@code Streamable} and runs them concurrently while
515+
* relaying inner items as first-come-first-served manner.
516+
* @param <R> the element type of the output sequence
517+
* @param mapper the function that turns an upstream item into a {@code Streamable} inner sequence
518+
* @param config the configuration record for this operator
519+
* @return the new {@code Streamable} instance
520+
* @throws NullPointerException if {@code mapper} or {@code config} is {@code null}
521+
*/
522+
@CheckReturnValue
523+
@NonNull
524+
default <R> Streamable<R> flatMap(@NonNull Function<? super T, ? extends Streamable<? extends R>> mapper,
525+
@NonNull StandardConcurrentConfig config) {
526+
Objects.requireNonNull(mapper, "mapper is null");
527+
Objects.requireNonNull(config, "config is null");
528+
return RxJavaPlugins.onAssembly(new StreamableFlatMap<>(this, mapper, config.maxConcurrency()));
529+
}
530+
472531
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
473532
// Consumption methods and outgoing converters
474533
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
@@ -525,17 +584,21 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
525584
Objects.requireNonNull(executor, "executor is null");
526585
final Streamable<T> me = this;
527586
var future = CompletableFuture.<Void>supplyAsync(() -> {
528-
try (var str = me.stream(canceller)) {
529-
while (!canceller.isDisposed()) {
530-
if (str.awaitNext(canceller)) {
531-
// System.out.println("Received " + str.current());
532-
consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"));
533-
} else {
534-
// System.out.println("EOF ");
535-
break;
587+
var str = me.stream(canceller);
588+
try {
589+
try {
590+
while (!canceller.isDisposed()) {
591+
if (str.awaitNext(canceller)) {
592+
// System.out.println("Received " + str.current());
593+
consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"));
594+
} else {
595+
// System.out.println("EOF ");
596+
break;
597+
}
536598
}
599+
} finally {
600+
str.awaitFinish(canceller);
537601
}
538-
// System.out.println("Canceller status after loop: " + canceller.isDisposed());
539602
} catch (final Throwable crash) {
540603
Exceptions.throwIfFatal(crash);
541604
if (crash instanceof CompletionException ce) {
@@ -567,19 +630,23 @@ default CompletionStageDisposable<Void> forEach(
567630
Objects.requireNonNull(executor, "executor is null");
568631
final Streamable<T> me = this;
569632
var future = CompletableFuture.<Void>supplyAsync(() -> {
570-
try (var str = me.stream(canceller)) {
633+
var str = me.stream(canceller);
634+
try {
635+
try {
571636
var stopper = Disposable.empty();
572-
while (!canceller.isDisposed() && !stopper.isDisposed()) {
573-
if (str.awaitNext(canceller)) {
574-
// System.out.println("Received " + str.current());
575-
var v = Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!");
576-
consumer.accept(v, stopper);
577-
} else {
578-
// System.out.println("EOF ");
579-
break;
637+
while (!canceller.isDisposed() && !stopper.isDisposed()) {
638+
if (str.awaitNext(canceller)) {
639+
// System.out.println("Received " + str.current());
640+
var v = Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!");
641+
consumer.accept(v, stopper);
642+
} else {
643+
// System.out.println("EOF ");
644+
break;
645+
}
580646
}
647+
} finally {
648+
str.awaitFinish(canceller);
581649
}
582-
// System.out.println("Canceller status after loop: " + canceller.isDisposed());
583650
} catch (final Throwable crash) {
584651
Exceptions.throwIfFatal(crash);
585652
if (crash instanceof CompletionException ce) {

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

Lines changed: 5 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
* TODO proper docs
3232
* @since 4.0.0
3333
*/
34-
public interface Streamer<@NonNull T> extends AutoCloseable, AwaitCoordinator {
34+
public interface Streamer<@NonNull T> extends AwaitCoordinator {
3535

3636
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
3737
// API
@@ -46,10 +46,11 @@ public interface Streamer<@NonNull T> extends AutoCloseable, AwaitCoordinator {
4646
CompletionStage<Boolean> next(@NonNull DisposableContainer cancellation);
4747

4848
/**
49-
* Returns the current element if {@link #next()} yielded {@code true}.
50-
* Can be called multiple times between {@link #next()} calls.
49+
* Returns the current element if {@link #next(DisposableContainer)} yielded {@code true}.
50+
* Can be called multiple times between {@link #next(DisposableContainer)} calls.
5151
* @return the current element
52-
* @throws NoSuchElementException before the very first {@link #next()} or after {@link #next()} returned {@code false}
52+
* @throws NoSuchElementException before the very first {@link #next(DisposableContainer)}
53+
* or after {@link #next(DisposableContainer)} returned {@code false}
5354
*/
5455
@NonNull
5556
T current();
@@ -67,80 +68,6 @@ public interface Streamer<@NonNull T> extends AutoCloseable, AwaitCoordinator {
6768
// HELPERS
6869
// oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
6970

70-
/**
71-
* Determine if there are more elements available from the source.
72-
* Uses a default, individual {@link CompositeDisposable} to manage cancellation.
73-
* @return eventually true or false, indicating availability or termination
74-
*/
75-
@NonNull
76-
default CompletionStage<Boolean> next() {
77-
return next(new CompositeDisposable());
78-
}
79-
80-
/**
81-
* Make this Streamer a resource and a Closeable, allowing virtually blocking closing.
82-
*/
83-
@Override
84-
default void close() {
85-
awaitFinish();
86-
}
87-
88-
/**
89-
* Augments the streamer so that the given canceller is injected into the various
90-
* lifecycle await calls.
91-
* @param canceller the canceller to inject
92-
* @return the augmented streamer
93-
*/
94-
default Streamer<T> finishVia(@NonNull DisposableContainer canceller) {
95-
Objects.requireNonNull(canceller, "canceller is null");
96-
if (this instanceof StreamerFinishViaDisposableContainerCanceller<T>(
97-
Streamer<T> streamer, DisposableContainer canceller1
98-
)) {
99-
if (streamer == this && canceller1 == canceller) {
100-
// DO not rewrap!
101-
return this;
102-
}
103-
}
104-
105-
return new StreamerFinishViaDisposableContainerCanceller<>(this, canceller);
106-
}
107-
108-
/**
109-
* Augments the base streamer with a canceller so that it can be injected at the various await calls.
110-
* @param <T> the element type of the stream
111-
*/
112-
static record StreamerFinishViaDisposableContainerCanceller<T>(
113-
@NonNull Streamer<T> streamer, @NonNull DisposableContainer canceller)
114-
implements Streamer<T> {
115-
116-
@Override
117-
public @NonNull CompletionStage<Boolean> next(@NonNull DisposableContainer cancellation) {
118-
// TODO Auto-generated method stub
119-
return streamer.next(cancellation);
120-
}
121-
122-
@Override
123-
public @NonNull T current() {
124-
return streamer.current();
125-
}
126-
127-
@Override
128-
public @NonNull CompletionStage<Void> finish(@NonNull DisposableContainer cancellation) {
129-
return streamer.finish(cancellation);
130-
}
131-
132-
}
133-
134-
/**
135-
* Moves and awaits the sequence's next element, returns false if there are no more
136-
* data.
137-
* @return true if the next element via {@link #current()} can be read, or false if
138-
* the stream ended.
139-
*/
140-
default boolean awaitNext() {
141-
return await(next());
142-
}
143-
14471
/**
14572
* Moves and awaits the sequence's next element, returns false if there are no more
14673
* data.
@@ -152,13 +79,6 @@ default boolean awaitNext(@NonNull DisposableContainer cancellation) {
15279
return await(next(cancellation), cancellation);
15380
}
15481

155-
/**
156-
* Finish and cleanup the sequence after its completion or cancellation.
157-
*/
158-
default void awaitFinish() {
159-
await(finish(DisposableContainer.NEVER), DisposableContainer.NEVER);
160-
}
161-
16282
/**
16383
* Who cancels the cancellation attempt? Another cancellation attempt!
16484
* @param cancellation the token to cancel and ongoing cancel attempt

‎src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFilter.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
/*
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava4.internal.operators.streamable;
15+
16+
import java.util.concurrent.*;
17+
import java.util.concurrent.atomic.AtomicInteger;
18+
19+
import io.reactivex.rxjava4.annotations.NonNull;
20+
import io.reactivex.rxjava4.core.*;
21+
import io.reactivex.rxjava4.disposables.DisposableContainer;
22+
import io.reactivex.rxjava4.exceptions.Exceptions;
23+
import io.reactivex.rxjava4.functions.Predicate;
24+
import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource;
25+
26+
public record StreamableFilter<T>(
27+
@NonNull Streamable<T> source,
28+
@NonNull Predicate<? super T> predicate)
29+
implements Streamable<T>, HasUpstreamStreamableSource<T> {
30+
31+
@Override
32+
public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) {
33+
return new FilterStreamer<>(source.stream(cancellation), predicate);
34+
}
35+
36+
static final class FilterStreamer<T> implements Streamer<T> {
37+
final Streamer<T> upstream;
38+
final Predicate<? super T> predicate;
39+
volatile T current;
40+
41+
final AtomicInteger wip = new AtomicInteger();
42+
43+
FilterStreamer(Streamer<T> upstream, Predicate<? super T> predicate) {
44+
this.upstream = upstream;
45+
this.predicate = predicate;
46+
}
47+
48+
@Override
49+
public @NonNull CompletionStage<Boolean> next(@NonNull DisposableContainer cancellation) {
50+
var cf = new CompletableFuture<Boolean>();
51+
drain(cf, cancellation);
52+
return cf;
53+
}
54+
55+
@Override
56+
public @NonNull T current() {
57+
return current;
58+
}
59+
60+
@Override
61+
public @NonNull CompletionStage<Void> finish(@NonNull DisposableContainer cancellation) {
62+
current = null;
63+
return upstream.finish(cancellation);
64+
}
65+
66+
void drain(CompletableFuture<Boolean> cf, DisposableContainer cancellation) {
67+
if (wip.getAndIncrement() != 0) {
68+
return;
69+
}
70+
do {
71+
upstream.next(cancellation)
72+
.whenComplete((v, e) -> {
73+
if (e != null) {
74+
cf.completeExceptionally(e);
75+
} else {
76+
if (v) {
77+
try {
78+
var w = upstream.current();
79+
if (predicate.test(w)) {
80+
current = w;
81+
cf.complete(true);
82+
} else {
83+
drain(cf, cancellation);
84+
}
85+
} catch (Throwable ex) {
86+
Exceptions.throwIfFatal(ex);
87+
cf.completeExceptionally(ex);
88+
}
89+
} else {
90+
cf.complete(false);
91+
}
92+
}
93+
});
94+
} while (wip.decrementAndGet() != 0);
95+
}
96+
}
97+
}

0 commit comments

Comments
 (0)