Skip to content

Commit 07ec596

Browse files
akarnokdakarnokd
committed
fix tests
1 parent 682143a commit 07ec596

10 files changed

Lines changed: 275 additions & 17 deletions

File tree

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

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -589,6 +589,17 @@ default Streamable<T> delay(long time, TimeUnit unit, Scheduler scheduler) {
589589
return RxJavaPlugins.onAssembly(new StreamableDelay<>(this, time, unit, scheduler));
590590
}
591591

592+
/**
593+
* Calls the specific {@link Consumer} if there is an error from the upstream.
594+
* @param consumer the consumer to call with the Throwable
595+
* @return the new {@code Streamable} instance
596+
* @throws NullPointerException if {@code consumer} is {@code null}
597+
*/
598+
default Streamable<T> doOnError(Consumer<? super Throwable> consumer) {
599+
Objects.requireNonNull(consumer, "consumer is null");
600+
return intercept(StreamableHelper.createOnError(consumer));
601+
}
602+
592603
/**
593604
* Calls the given consumer whenever an upstream item becomes available.
594605
* @param consumer the callback to invoke with the next item from upstream
@@ -983,6 +994,27 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
983994
subscribe(subscriber, Executors.newVirtualThreadPerTaskExecutor());
984995
}
985996

997+
/**
998+
* Relays the events of the upstream into a {@link StreamerInput} consumer.
999+
* @param consumer the consumer to relay events into
1000+
* @throws NullPointerException if {@code consumer} or {@code executor} is {@code null}
1001+
*/
1002+
default void subscribe(@NonNull StreamerInput<? super T> consumer) {
1003+
subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor());
1004+
}
1005+
1006+
/**
1007+
* Relays the events of the upstream into a {@link StreamerInput} consumer.
1008+
* @param consumer the consumer to relay events into
1009+
* @param executor the {@link ExecutorService} to run the blocking consume and emissions
1010+
* @throws NullPointerException if {@code consumer} or {@code executor} is {@code null}
1011+
*/
1012+
default void subscribe(@NonNull StreamerInput<? super T> consumer, ExecutorService executor) {
1013+
Objects.requireNonNull(consumer, "consumer is null");
1014+
Objects.requireNonNull(executor, "executor is null");
1015+
StreamableForEach.forEach(this, consumer, executor);
1016+
}
1017+
9861018
/**
9871019
* Creates a new {@link TestSubscriber} and subscribes it to this {@code Streamable}.
9881020
* @return the created test subscriber

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

Lines changed: 28 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313

1414
package io.reactivex.rxjava4.core;
1515

16+
import java.util.NoSuchElementException;
1617
import java.util.concurrent.*;
1718

1819
import io.reactivex.rxjava4.annotations.NonNull;
@@ -50,6 +51,7 @@ public interface Streamer<@NonNull T> {
5051
* Calling it during an ongoing [#next()] or [#finish()] call, or beyond the lifecycle of the `Streamer`
5152
* is an undefined behavior. It may yield `null` or throw.
5253
* @return the current item
54+
* @throws NoSuchElementException if there are no items to return
5355
*/
5456
@NonNull
5557
T current();
@@ -76,25 +78,42 @@ public interface Streamer<@NonNull T> {
7678
* @return true if there are more items, false if no more items are coming, or crashes
7779
*/
7880
default boolean awaitNext() {
79-
var s = next();
80-
if (s == NEXT_TRUE) {
81+
return awaitBoolean(next());
82+
}
83+
84+
/**
85+
* Convenience method to blockingly await the CompletionStage returned by the {@link #finish()} method.
86+
*/
87+
default void awaitFinish() {
88+
awaitVoid(finish());
89+
}
90+
91+
/**
92+
* Convenience method to await the completion of a boolean stage, optimized
93+
* for handling {@value #NEXT_TRUE} and {@value #NEXT_FALSE} directly.
94+
* @param stage the stage to await
95+
* @return the result of the stage
96+
*/
97+
static boolean awaitBoolean(CompletionStage<Boolean> stage) {
98+
if (stage == NEXT_TRUE) {
8199
return true;
82100
} else
83-
if (s == NEXT_FALSE) {
101+
if (stage == NEXT_FALSE) {
84102
return false;
85103
}
86-
return s.toCompletableFuture().join();
104+
return stage.toCompletableFuture().join();
87105
}
88106

89107
/**
90-
* Convenience method to blockingly await the CompletionStage returned by the {@link #finish()} method.
108+
* Convenience method to await the completion of a stage, optimized
109+
* for handling {@value #FINISHED} directly.
110+
* @param stage the stage to await
91111
*/
92-
default void awaitFinish() {
93-
var s = finish();
94-
if (s == FINISHED) {
112+
static void awaitVoid(CompletionStage<Void> stage) {
113+
if (stage == FINISHED) {
95114
return;
96115
}
97-
s.toCompletableFuture().join();
116+
stage.toCompletableFuture().join();
98117
}
99118

100119
/**

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import java.util.concurrent.Flow.Subscriber;
1818

1919
import io.reactivex.rxjava4.annotations.*;
20+
import io.reactivex.rxjava4.disposables.*;
2021

2122
/**
2223
* An interface to submit items and terminal events to a consumer that indacates when the processing of
@@ -45,4 +46,13 @@ public interface StreamerInput<@NonNull T> {
4546
* or exceptionally on error
4647
*/
4748
CompletionStage<Void> finish(@Nullable Throwable throwable);
49+
50+
/**
51+
* Returns the {@link DisposableContainer} to use to detect if the consumer has indicated no more
52+
* items it is willing to accept.
53+
* @return the {@code DisposableContainer}
54+
*/
55+
default DisposableContainer cancellation() {
56+
return new CompositeDisposable();
57+
}
4858
}

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

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import io.reactivex.rxjava4.exceptions.Exceptions;
2323
import io.reactivex.rxjava4.functions.*;
2424
import io.reactivex.rxjava4.internal.util.ExceptionHelper;
25+
import io.reactivex.rxjava4.plugins.RxJavaPlugins;
2526

2627
/**
2728
* ForEach implementation to unclutter the {@link Streamable} type.
@@ -109,4 +110,43 @@ public static <T> CompletionStageDisposable<Void> forEach(
109110
canceller.add(Disposable.fromFuture(future));
110111
return new CompletionStageDisposable<>(future, canceller);
111112
}
113+
114+
public static <T> void forEach(Streamable<T> me, StreamerInput<? super T> consumer, ExecutorService executor) {
115+
CompletableFuture.runAsync(() -> {
116+
Throwable error = null;
117+
var cancellation = consumer.cancellation();
118+
var streamer = me.stream(cancellation);
119+
try {
120+
try {
121+
while (!cancellation.isDisposed()) {
122+
if (streamer.awaitNext()) {
123+
Streamer.awaitBoolean(consumer.next(streamer.current()));
124+
} else {
125+
break;
126+
}
127+
}
128+
} finally {
129+
try {
130+
streamer.awaitFinish();
131+
} catch (Throwable ex) {
132+
Exceptions.throwIfFatal(ex);
133+
error = ExceptionHelper.unwrap(ex);
134+
}
135+
}
136+
} catch (Throwable crash) {
137+
Exceptions.throwIfFatal(crash);
138+
crash = ExceptionHelper.unwrap(crash);
139+
if (error != null) {
140+
crash.addSuppressed(error);
141+
}
142+
error = crash;
143+
}
144+
try {
145+
Streamer.awaitVoid(consumer.finish(error));
146+
} catch (Throwable ex) {
147+
Exceptions.throwIfFatal(ex);
148+
RxJavaPlugins.onError(ex);
149+
}
150+
}, executor);
151+
}
112152
}

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

Lines changed: 30 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,10 @@
2121

2222
import io.reactivex.rxjava4.annotations.*;
2323
import io.reactivex.rxjava4.core.*;
24+
import io.reactivex.rxjava4.core.config.StreamableInterceptConfig;
2425
import io.reactivex.rxjava4.disposables.Disposable;
25-
import io.reactivex.rxjava4.exceptions.CompositeException;
26+
import io.reactivex.rxjava4.exceptions.*;
27+
import io.reactivex.rxjava4.functions.Consumer;
2628
import io.reactivex.rxjava4.internal.util.*;
2729

2830
/**
@@ -373,4 +375,30 @@ public void accept(Boolean t, Throwable u) {
373375
}
374376
}
375377
}
376-
}
378+
379+
/**
380+
* Create a {@link StreamableInterceptConfig} that can consume the {@link Streamer#next()} errors.
381+
* @param <T> the element type of the {@link Streamable}
382+
* @param consumer the consumer to be called with the error
383+
* @return the new {@code StreamableInterceptConfig} instance
384+
*/
385+
public static <T> StreamableInterceptConfig<T> createOnError(Consumer<? super Throwable> consumer) {
386+
return new StreamableInterceptConfig<>((_, v) -> v, (_, v) -> {
387+
var cf = new CompletableFuture<Boolean>();
388+
v.whenComplete((u, e) -> {
389+
if (e != null) {
390+
try {
391+
consumer.accept(e);
392+
} catch (Throwable ex) {
393+
Exceptions.throwIfFatal(ex);
394+
ex.addSuppressed(e);
395+
e = ex;
396+
}
397+
cf.completeExceptionally(e);
398+
} else {
399+
cf.complete(u);
400+
}
401+
});
402+
return cf;
403+
}, v -> v, (_, v) -> v);
404+
}}

‎src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnErrorTest.java‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
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 static org.junit.jupiter.api.Assertions.*;
17+
18+
import java.io.IOException;
19+
import java.util.concurrent.TimeUnit;
20+
import java.util.concurrent.atomic.AtomicReference;
21+
22+
import org.junit.jupiter.api.Test;
23+
24+
import io.reactivex.rxjava4.core.Streamable;
25+
import io.reactivex.rxjava4.exceptions.TestException;
26+
27+
public class StreamableDoOnErrorTest extends StreamableBaseTest {
28+
29+
@Test
30+
public void normal() {
31+
AtomicReference<Throwable> error = new AtomicReference<>();
32+
Streamable.range(1, 5)
33+
.doOnError(error::set)
34+
.test()
35+
.awaitDone(5, TimeUnit.SECONDS)
36+
.assertResult(1, 2, 3, 4, 5)
37+
;
38+
39+
assertNull(error.get(), "error is not empty?");
40+
}
41+
42+
@Test
43+
public void hasError() {
44+
AtomicReference<Throwable> error = new AtomicReference<>();
45+
var te = new TestException();
46+
Streamable.error(te)
47+
.doOnError(error::set)
48+
.test()
49+
.awaitDone(5, TimeUnit.SECONDS)
50+
.assertFailure(TestException.class)
51+
;
52+
53+
assertSame(te, error.get(), "doOnError differs from TestSubscriber.onError?");
54+
}
55+
56+
@Test
57+
public void consumerCrash() {
58+
var te = new TestException();
59+
Streamable.error(te)
60+
.doOnError(_ -> { throw new IOException(); })
61+
.test()
62+
.awaitDone(5, TimeUnit.SECONDS)
63+
.assertFailure(IOException.class)
64+
.assertError(e -> e.getSuppressed()[0] == te)
65+
;
66+
}
67+
}

‎src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
import io.reactivex.rxjava4.core.Streamable;
2424
import io.reactivex.rxjava4.exceptions.TestException;
2525

26-
public class StreamableDoOnXTest extends StreamableBaseTest {
26+
public class StreamableDoOnNextTest extends StreamableBaseTest {
2727

2828
@Test
2929
public void passthrough() {

‎src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java‎

Lines changed: 60 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,15 @@
1515

1616
import static org.junit.jupiter.api.Assertions.*;
1717

18-
import java.util.concurrent.CancellationException;
18+
import java.util.concurrent.*;
1919
import java.util.concurrent.atomic.AtomicInteger;
2020

21-
import org.junit.jupiter.api.*;
21+
import org.junit.jupiter.api.Test;
22+
2223
import io.reactivex.rxjava4.core.Streamable;
2324
import io.reactivex.rxjava4.disposables.CompositeDisposable;
2425
import io.reactivex.rxjava4.exceptions.*;
26+
import io.reactivex.rxjava4.processors.DispatchStreamProcessor;
2527

2628
public class StreamableForEachTest extends StreamableBaseTest {
2729

@@ -189,4 +191,60 @@ public void forEachBiInsideCancel() throws Throwable {
189191
assertEquals(1, counter.get());
190192
});
191193
}
194+
195+
@Test
196+
public void forEachInput() throws Throwable {
197+
var dsp = new DispatchStreamProcessor<>();
198+
var ts = dsp.test();
199+
200+
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
201+
202+
while (!dsp.hasStreamers()) {
203+
Thread.sleep(0, 1000);
204+
}
205+
206+
Streamable.range(1, 5)
207+
.subscribe(dsp);
208+
209+
ts.awaitDone(5, TimeUnit.SECONDS)
210+
.assertResult(1, 2, 3, 4, 5);
211+
}
212+
213+
@Test
214+
public void forEachInputDebug() throws Throwable {
215+
withCachedExecutor(exec -> {
216+
var dsp = new DispatchStreamProcessor<>();
217+
var ts = dsp.test(exec);
218+
219+
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
220+
221+
while (!dsp.hasStreamers()) {
222+
Thread.sleep(0, 1000);
223+
}
224+
225+
Streamable.range(1, 5)
226+
.subscribe(dsp);
227+
228+
ts.awaitDone(5, TimeUnit.SECONDS)
229+
.assertResult(1, 2, 3, 4, 5);
230+
});
231+
}
232+
233+
@Test
234+
public void forEachInputError() throws Throwable {
235+
var dsp = new DispatchStreamProcessor<>();
236+
var ts = dsp.test();
237+
238+
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
239+
240+
while (!dsp.hasStreamers()) {
241+
Thread.sleep(0, 1000);
242+
}
243+
244+
Streamable.error(new TestException())
245+
.subscribe(dsp);
246+
247+
ts.awaitDone(5, TimeUnit.SECONDS)
248+
.assertFailure(TestException.class);
249+
}
192250
}

0 commit comments

Comments
 (0)