Skip to content

4.x: Unit test lambdaification 12 of N #8150

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
Jun 22, 2026
+620 −1,257
Merged

4.x: Unit test lambdaification 12 of N #8150

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
121 changes: 46 additions & 75 deletions ...st/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableMergeDelayErrorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@

import io.reactivex.rxjava4.core.*;
import io.reactivex.rxjava4.exceptions.*;
import io.reactivex.rxjava4.functions.LongConsumer;
import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription;
import io.reactivex.rxjava4.processors.PublishProcessor;
import io.reactivex.rxjava4.subscribers.*;
Expand Down Expand Up @@ -224,18 +223,13 @@ public void mergeFlowableOfFlowables() {
final Flowable<String> f1 = Flowable.unsafeCreate(new TestSynchronousFlowable());
final Flowable<String> f2 = Flowable.unsafeCreate(new TestSynchronousFlowable());

Flowable<Flowable<String>> flowableOfFlowables = Flowable.unsafeCreate(new Publisher<Flowable<String>>() {

@Override
public void subscribe(Subscriber<? super Flowable<String>> subscriber) {
subscriber.onSubscribe(new BooleanSubscription());
// simulate what would happen in a Flowable
subscriber.onNext(f1);
subscriber.onNext(f2);
subscriber.onComplete();
}

});
Flowable<Flowable<String>> flowableOfFlowables = Flowable.unsafeCreate(subscriber -> {
subscriber.onSubscribe(new BooleanSubscription());
// simulate what would happen in a Flowable
subscriber.onNext(f1);
subscriber.onNext(f2);
subscriber.onComplete();
});
Flowable<String> m = Flowable.mergeDelayError(flowableOfFlowables);
m.subscribe(stringSubscriber);

Expand Down Expand Up @@ -298,7 +292,7 @@ public void synchronousError() {
final Flowable<Flowable<String>> f1 = Flowable.error(new RuntimeException("unit test"));

final CountDownLatch latch = new CountDownLatch(1);
Flowable.mergeDelayError(f1).subscribe(new DefaultSubscriber<String>() {
Flowable.mergeDelayError(f1).subscribe(new DefaultSubscriber<String>() /* NFI */ {
@Override
public void onComplete() {
fail("Expected onError path");
Expand Down Expand Up @@ -338,15 +332,10 @@ private static class TestASynchronousFlowable implements Publisher<String> {
@Override
public void subscribe(final Subscriber<? super String> subscriber) {
subscriber.onSubscribe(new BooleanSubscription());
t = new Thread(new Runnable() {

@Override
public void run() {
subscriber.onNext("hello");
subscriber.onComplete();
}

});
t = new Thread(() -> {
subscriber.onNext("hello");
subscriber.onComplete();
});
t.start();
}
}
Expand Down Expand Up @@ -393,29 +382,24 @@ private static class TestAsyncErrorFlowable implements Publisher<String> {
@Override
public void subscribe(final Subscriber<? super String> subscriber) {
subscriber.onSubscribe(new BooleanSubscription());
t = new Thread(new Runnable() {

@Override
public void run() {
for (String s : valuesToReturn) {
if (s == null) {
System.out.println("throwing exception");
try {
Thread.sleep(100);
} catch (Throwable e) {

}
subscriber.onError(new NullPointerException());
return;
} else {
subscriber.onNext(s);
}
}
System.out.println("subscription complete");
subscriber.onComplete();
}

});
t = new Thread(() -> {
for (String s : valuesToReturn) {
if (s == null) {
System.out.println("throwing exception");
try {
Thread.sleep(100);
} catch (Throwable e) {

}
subscriber.onError(new NullPointerException());
return;
} else {
subscriber.onNext(s);
}
}
System.out.println("subscription complete");
subscriber.onComplete();
});
t.start();
}
}
Expand Down Expand Up @@ -459,15 +443,12 @@ public void errorInParentFlowableDelayed() throws Exception {
for (int i = 0; i < 50; i++) {
final TestASynchronous1sDelayedFlowable f1 = new TestASynchronous1sDelayedFlowable();
final TestASynchronous1sDelayedFlowable f2 = new TestASynchronous1sDelayedFlowable();
Flowable<Flowable<String>> parentFlowable = Flowable.unsafeCreate(new Publisher<Flowable<String>>() {
@Override
public void subscribe(Subscriber<? super Flowable<String>> op) {
op.onSubscribe(new BooleanSubscription());
op.onNext(Flowable.unsafeCreate(f1));
op.onNext(Flowable.unsafeCreate(f2));
op.onError(new NullPointerException("throwing exception in parent"));
}
});
Flowable<Flowable<String>> parentFlowable = Flowable.unsafeCreate(op -> {
op.onSubscribe(new BooleanSubscription());
op.onNext(Flowable.unsafeCreate(f1));
op.onNext(Flowable.unsafeCreate(f2));
op.onError(new NullPointerException("throwing exception in parent"));
});

stringSubscriber = TestHelper.mockSubscriber();

Expand All @@ -490,20 +471,15 @@ private static class TestASynchronous1sDelayedFlowable implements Publisher<Stri
@Override
public void subscribe(final Subscriber<? super String> subscriber) {
subscriber.onSubscribe(new BooleanSubscription());
t = new Thread(new Runnable() {

@Override
public void run() {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
subscriber.onError(e);
}
subscriber.onNext("hello");
subscriber.onComplete();
}

});
t = new Thread(() -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
subscriber.onError(e);
}
subscriber.onNext("hello");
subscriber.onComplete();
});
t.start();
}
}
Expand All @@ -514,12 +490,7 @@ public void delayErrorMaxConcurrent() {
Flowable<Integer> source = Flowable.mergeDelayError(Flowable.just(
Flowable.just(1).hide(),
Flowable.<Integer>error(new TestException()))
.doOnRequest(new LongConsumer() {
@Override
public void accept(long t1) {
requests.add(t1);
}
}), 1);
.doOnRequest(t1 -> requests.add(t1)), 1);

TestSubscriberEx<Integer> ts = new TestSubscriberEx<>();

Expand Down
35 changes: 15 additions & 20 deletions ...java/io/reactivex/rxjava4/internal/operators/flowable/FlowableMergeMaxConcurrentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,25 +92,20 @@ private static class SubscriptionCheckObservable implements Publisher<String> {
@Override
public void subscribe(final Subscriber<? super String> t1) {
t1.onSubscribe(new BooleanSubscription());
new Thread(new Runnable() {

@Override
public void run() {
if (subscriptionCount.incrementAndGet() > maxConcurrent) {
failed = true;
}
t1.onNext("one");
t1.onNext("two");
t1.onNext("three");
t1.onNext("four");
t1.onNext("five");
// We could not decrement subscriptionCount in the unsubscribe method
// as "unsubscribe" is not guaranteed to be called before the next "subscribe".
subscriptionCount.decrementAndGet();
t1.onComplete();
}

}).start();
new Thread(() -> {
if (subscriptionCount.incrementAndGet() > maxConcurrent) {
failed = true;
}
t1.onNext("one");
t1.onNext("two");
t1.onNext("three");
t1.onNext("four");
t1.onNext("five");
// We could not decrement subscriptionCount in the unsubscribe method
// as "unsubscribe" is not guaranteed to be called before the next "subscribe".
subscriptionCount.decrementAndGet();
t1.onComplete();
}).start();
}

}
Expand Down Expand Up @@ -261,7 +256,7 @@ public void backpressureHonored() throws Exception {

final CountDownLatch cdl = new CountDownLatch(5);

TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L) {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L) /* NFI */ {
@Override
public void onNext(Integer t) {
super.onNext(t);
Expand Down
Loading