From 67e915b9db6c1baadb92773de82421d4a98ef9c7 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Sun, 21 Jun 2026 14:23:58 +0200 Subject: [PATCH] 4.x: Unit test lambdaification 6 of N --- .../reactivex/rxjava4/core/ConverterTest.java | 2 +- .../flowable/FlowableCovarianceTest.java | 2 - .../rxjava4/flowable/FlowableDoOnTest.java | 1 - .../flowable/FlowableGroupByTests.java | 1 - .../rxjava4/flowable/FlowableMergeTests.java | 1 - .../rxjava4/flowable/FlowableNullTests.java | 1 - .../flowable/FlowableSubscriberTest.java | 3 +- .../rxjava4/flowable/FlowableZipTests.java | 1 - .../jdk8/ParallelMapOptionalTest.java | 1 - .../observers/LambdaObserverTest.java | 1 - .../observers/ObservableConsumersTest.java | 23 +-- .../observers/QueueDrainObserverTest.java | 14 +- .../observers/SingleConsumersTest.java | 16 +- .../completable/CompletableAmbTest.java | 54 ++--- .../CompletableAndThenCompletableTest.java | 44 +---- .../completable/CompletableCacheTest.java | 34 +--- .../completable/CompletableConcatTest.java | 99 +++------- .../completable/CompletableCreateTest.java | 187 +++++++----------- .../CompletableDelaySubscriptionTest.java | 40 +--- .../completable/CompletableDelayTest.java | 16 +- .../completable/CompletableDetachTest.java | 14 +- .../completable/CompletableDisposeOnTest.java | 15 +- .../completable/CompletableDoFinallyTest.java | 15 +- .../CompletableDoOnLifecycleTest.java | 2 +- .../completable/CompletableDoOnTest.java | 31 +-- .../flowable/FlowableGroupByTest.java | 4 +- 26 files changed, 175 insertions(+), 447 deletions(-) diff --git a/src/test/java/io/reactivex/rxjava4/core/ConverterTest.java b/src/test/java/io/reactivex/rxjava4/core/ConverterTest.java index 3594d55e3d..a1d7a0a104 100644 --- a/src/test/java/io/reactivex/rxjava4/core/ConverterTest.java +++ b/src/test/java/io/reactivex/rxjava4/core/ConverterTest.java @@ -194,7 +194,7 @@ interface B { } } private static ParallelFlowableConverter, B> testParallelFlowableConverterCreator() { - return a -> new B() /* NFI */ { + return _ -> new B() /* NFI */ { }; } diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java index 413482ea82..89b1398b09 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java @@ -19,10 +19,8 @@ import java.util.concurrent.Flow.Publisher; import org.junit.Test; -import static java.util.concurrent.Flow.*; import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.flowable.FlowableCovarianceTest.Movie; import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.testsupport.TestSubscriberEx; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableDoOnTest.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableDoOnTest.java index efc88f5946..84bdf7575b 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableDoOnTest.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableDoOnTest.java @@ -21,7 +21,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.*; public class FlowableDoOnTest extends RxJavaTest { diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java index 8f7ca9616c..4d6e145ee7 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java @@ -14,7 +14,6 @@ package io.reactivex.rxjava4.flowable; import org.junit.Test; -import static java.util.concurrent.Flow.*; import java.util.concurrent.Flow.Publisher; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableMergeTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableMergeTests.java index 241bbdf7a2..2b7ff264d1 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableMergeTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableMergeTests.java @@ -19,7 +19,6 @@ import java.util.concurrent.Flow.Publisher; import org.junit.Test; -import static java.util.concurrent.Flow.*; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.flowable.FlowableCovarianceTest.*; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableNullTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableNullTests.java index 6b18a86a88..cc13adf285 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableNullTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableNullTests.java @@ -21,7 +21,6 @@ import java.util.concurrent.Flow.Publisher; import org.junit.*; -import static java.util.concurrent.Flow.*; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.*; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableSubscriberTest.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableSubscriberTest.java index 3e78a2d27e..dd42979b94 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableSubscriberTest.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableSubscriberTest.java @@ -24,7 +24,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.subscribers.ForEachWhileSubscriber; import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription; @@ -616,7 +615,7 @@ public void onCompleteThrows() { public void subscribeConsumerConsumerWithError() { final List list = new ArrayList<>(); - Flowable.error(new TestException()).subscribe(v -> list.add(v), e -> list.add(100)); + Flowable.error(new TestException()).subscribe(v -> list.add(v), _ -> list.add(100)); assertEquals(Arrays.asList(100), list); } diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java index 2588fbf7ad..238443fd0c 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java @@ -19,7 +19,6 @@ import java.util.concurrent.Flow.Publisher; import org.junit.Test; -import static java.util.concurrent.Flow.*; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.flowable.FlowableCovarianceTest.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/jdk8/ParallelMapOptionalTest.java b/src/test/java/io/reactivex/rxjava4/internal/jdk8/ParallelMapOptionalTest.java index 75a2598bc4..3f6a1a1cdb 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/jdk8/ParallelMapOptionalTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/jdk8/ParallelMapOptionalTest.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.parallel.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; diff --git a/src/test/java/io/reactivex/rxjava4/internal/observers/LambdaObserverTest.java b/src/test/java/io/reactivex/rxjava4/internal/observers/LambdaObserverTest.java index c184bdb2da..eebf15b203 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/observers/LambdaObserverTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/observers/LambdaObserverTest.java @@ -25,7 +25,6 @@ import io.reactivex.rxjava4.core.RxJavaTest; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.subjects.PublishSubject; diff --git a/src/test/java/io/reactivex/rxjava4/internal/observers/ObservableConsumersTest.java b/src/test/java/io/reactivex/rxjava4/internal/observers/ObservableConsumersTest.java index 896f8b1348..015d5e565f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/observers/ObservableConsumersTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/observers/ObservableConsumersTest.java @@ -211,11 +211,8 @@ public void onCompleteDispose() { public void onNextCrash() { List errors = TestHelper.trackPluginErrors(); try { - subscribeAutoDispose(processor, composite, new Consumer() { - @Override - public void accept(Object t) throws Exception { - throw new IOException(); - } + subscribeAutoDispose(processor, composite, _ -> { + throw new IOException(); }, this, this); processor.onNext(1); @@ -232,11 +229,8 @@ public void accept(Object t) throws Exception { public void onNextCrashOnError() { List errors = TestHelper.trackPluginErrors(); try { - subscribeAutoDispose(processor, composite, this, new Consumer() { - @Override - public void accept(Throwable t) throws Exception { - throw new IOException(t); - } + subscribeAutoDispose(processor, composite, this, t -> { + throw new IOException(t); }, this); processor.onError(new IllegalArgumentException()); @@ -275,11 +269,8 @@ public void onNextCrashNoError() { public void onCompleteCrash() { List errors = TestHelper.trackPluginErrors(); try { - subscribeAutoDispose(processor, composite, this, this, new Action() { - @Override - public void run() throws Exception { - throw new IOException(); - } + subscribeAutoDispose(processor, composite, this, this, () -> { + throw new IOException(); }); processor.onNext(1); @@ -298,7 +289,7 @@ public void badSource() { List errors = TestHelper.trackPluginErrors(); try { subscribeAutoDispose( - new Observable() { + new Observable() /* NFI */ { @Override protected void subscribeActual( Observer observer) { diff --git a/src/test/java/io/reactivex/rxjava4/internal/observers/QueueDrainObserverTest.java b/src/test/java/io/reactivex/rxjava4/internal/observers/QueueDrainObserverTest.java index 6a32cc2380..1ad18ab285 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/observers/QueueDrainObserverTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/observers/QueueDrainObserverTest.java @@ -125,12 +125,7 @@ public void unorderedOnNextRace() { final QueueDrainObserver qd = createUnordered(to, d); to.onSubscribe(Disposable.empty()); - Runnable r1 = new Runnable() { - @Override - public void run() { - qd.onNext(1); - } - }; + Runnable r1 = () -> qd.onNext(1); TestHelper.race(r1, r1); @@ -147,12 +142,7 @@ public void orderedOnNextRace() { final QueueDrainObserver qd = createOrdered(to, d); to.onSubscribe(Disposable.empty()); - Runnable r1 = new Runnable() { - @Override - public void run() { - qd.onNext(1); - } - }; + Runnable r1 = () -> qd.onNext(1); TestHelper.race(r1, r1); diff --git a/src/test/java/io/reactivex/rxjava4/internal/observers/SingleConsumersTest.java b/src/test/java/io/reactivex/rxjava4/internal/observers/SingleConsumersTest.java index 5e69a873bb..f2f53bac7f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/observers/SingleConsumersTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/observers/SingleConsumersTest.java @@ -122,11 +122,8 @@ public void onErrorError() { public void onSuccessCrash() { List errors = TestHelper.trackPluginErrors(); try { - subscribeAutoDispose(processor, composite, new Consumer() { - @Override - public void accept(Object t) throws Exception { - throw new IOException(); - } + subscribeAutoDispose(processor, composite, _ -> { + throw new IOException(); }, this); processor.onSuccess(1); @@ -143,11 +140,8 @@ public void accept(Object t) throws Exception { public void onErrorCrash() { List errors = TestHelper.trackPluginErrors(); try { - subscribeAutoDispose(processor, composite, this, new Consumer() { - @Override - public void accept(Throwable t) throws Exception { - throw new IOException(t); - } + subscribeAutoDispose(processor, composite, this, t -> { + throw new IOException(t); }); processor.onError(new IllegalArgumentException()); @@ -168,7 +162,7 @@ public void badSource() { List errors = TestHelper.trackPluginErrors(); try { subscribeAutoDispose( - new Single() { + new Single() /* NFI */ { @Override protected void subscribeActual( SingleObserver observer) { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAmbTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAmbTest.java index eb5eea9209..b27f108b5e 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAmbTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAmbTest.java @@ -24,7 +24,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.operators.completable.CompletableAmb.Amb; import io.reactivex.rxjava4.observers.TestObserver; @@ -88,19 +87,9 @@ public void innerErrorRace() { final TestException ex = new TestException(); - Runnable r1 = new Runnable() { - @Override - public void run() { - pp0.onError(ex); - } - }; + Runnable r1 = () -> pp0.onError(ex); - Runnable r2 = new Runnable() { - @Override - public void run() { - pp1.onError(ex); - } - }; + Runnable r2 = () -> pp1.onError(ex); TestHelper.race(r1, r2); @@ -127,19 +116,9 @@ public void nullSourceSuccessRace() { final Completable source = Completable.ambArray(ps.ignoreElements(), Completable.never(), Completable.never(), null); - Runnable r1 = new Runnable() { - @Override - public void run() { - source.test(); - } - }; + Runnable r1 = () -> source.test(); - Runnable r2 = new Runnable() { - @Override - public void run() { - ps.onComplete(); - } - }; + Runnable r2 = () -> ps.onComplete(); TestHelper.race(r1, r2); @@ -278,12 +257,9 @@ public void noWinnerErrorDispose() throws Exception { .observeOn(Schedulers.computation()), Completable.never() ) - .subscribe(Functions.EMPTY_ACTION, new Consumer() { - @Override - public void accept(Throwable e) throws Exception { - interrupted.set(Thread.currentThread().isInterrupted()); - cdl.countDown(); - } + .subscribe(Functions.EMPTY_ACTION, _ -> { + interrupted.set(Thread.currentThread().isInterrupted()); + cdl.countDown(); }); assertTrue(cdl.await(500, TimeUnit.SECONDS)); @@ -303,12 +279,9 @@ public void noWinnerCompleteDispose() throws Exception { .observeOn(Schedulers.computation()), Completable.never() ) - .subscribe(new Action() { - @Override - public void run() throws Exception { - interrupted.set(Thread.currentThread().isInterrupted()); - cdl.countDown(); - } + .subscribe(() -> { + interrupted.set(Thread.currentThread().isInterrupted()); + cdl.countDown(); }); assertTrue(cdl.await(500, TimeUnit.SECONDS)); @@ -318,12 +291,7 @@ public void run() throws Exception { @Test public void completableSourcesInIterable() { - CompletableSource source = new CompletableSource() { - @Override - public void subscribe(CompletableObserver observer) { - Completable.complete().subscribe(observer); - } - }; + CompletableSource source = observer -> Completable.complete().subscribe(observer); Completable.amb(Arrays.asList(source, source)) .test() diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAndThenCompletableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAndThenCompletableTest.java index a713fc9961..163e4d407a 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAndThenCompletableTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableAndThenCompletableTest.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.Action; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.testsupport.TestHelper; @@ -93,12 +92,7 @@ public void andThenCompletableErrorError() { @Test public void andThenCanceled() { final AtomicInteger completableRunCount = new AtomicInteger(); - Completable.fromRunnable(new Runnable() { - @Override - public void run() { - completableRunCount.incrementAndGet(); - } - }) + Completable.fromRunnable(() -> completableRunCount.incrementAndGet()) .andThen(Completable.complete()) .test(true) .assertEmpty(); @@ -108,12 +102,7 @@ public void run() { @Test public void andThenFirstCancels() { final TestObserver to = new TestObserver<>(); - Completable.fromRunnable(new Runnable() { - @Override - public void run() { - to.dispose(); - } - }) + Completable.fromRunnable(() -> to.dispose()) .andThen(Completable.complete()) .subscribe(to); to @@ -125,12 +114,7 @@ public void run() { public void andThenSecondCancels() { final TestObserver to = new TestObserver<>(); Completable.complete() - .andThen(Completable.fromRunnable(new Runnable() { - @Override - public void run() { - to.dispose(); - } - })) + .andThen(Completable.fromRunnable(() -> to.dispose())) .subscribe(to); to .assertNotComplete() @@ -154,23 +138,15 @@ public void andThenNoInterrupt() throws InterruptedException { Completable.complete() .subscribeOn(Schedulers.cached()) .observeOn(Schedulers.cached()) - .andThen(Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - try { - Thread.sleep(30); - } catch (InterruptedException e) { - System.out.println("Interrupted! " + Thread.currentThread()); - interrupted[0] = true; - } + .andThen(Completable.fromAction(() -> { + try { + Thread.sleep(30); + } catch (InterruptedException e) { + System.out.println("Interrupted! " + Thread.currentThread()); + interrupted[0] = true; } })) - .subscribe(new Action() { - @Override - public void run() throws Exception { - latch.countDown(); - } - }); + .subscribe(() -> latch.countDown()); } latch.await(); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCacheTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCacheTest.java index f3f2ed47fb..5eeaf9aabb 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCacheTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCacheTest.java @@ -88,7 +88,7 @@ public void crossDispose() { final TestObserver to1 = new TestObserver<>(); - final TestObserver to2 = new TestObserver() { + final TestObserver to2 = new TestObserver() /* NFI */ { @Override public void onComplete() { super.onComplete(); @@ -113,7 +113,7 @@ public void crossDisposeOnError() { final TestObserver to1 = new TestObserver<>(); - final TestObserver to2 = new TestObserver() { + final TestObserver to2 = new TestObserver() /* NFI */ { @Override public void onError(Throwable ex) { super.onError(ex); @@ -178,18 +178,8 @@ public void subscribeRace() { final TestObserver to2 = new TestObserver<>(); - Runnable r1 = new Runnable() { - @Override - public void run() { - c.subscribe(to1); - } - }; - Runnable r2 = new Runnable() { - @Override - public void run() { - c.subscribe(to2); - } - }; + Runnable r1 = () -> c.subscribe(to1); + Runnable r2 = () -> c.subscribe(to2); TestHelper.race(r1, r2); @@ -211,18 +201,8 @@ public void subscribeDisposeRace() { final TestObserver to2 = new TestObserver<>(); - Runnable r1 = new Runnable() { - @Override - public void run() { - to1.dispose(); - } - }; - Runnable r2 = new Runnable() { - @Override - public void run() { - c.subscribe(to2); - } - }; + Runnable r1 = () -> to1.dispose(); + Runnable r2 = () -> c.subscribe(to2); TestHelper.race(r1, r2); @@ -241,7 +221,7 @@ public void doubleDispose() { final TestObserver to = new TestObserver<>(); ps.ignoreElements().cache() - .subscribe(new CompletableObserver() { + .subscribe(new CompletableObserver() /* NFI */ { @Override public void onSubscribe(Disposable d) { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableConcatTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableConcatTest.java index 026785fd66..ca1a51d11e 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableConcatTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableConcatTest.java @@ -39,16 +39,13 @@ public void overflowReported() { List errors = TestHelper.trackPluginErrors(); try { Completable.concat( - Flowable.fromPublisher(new Publisher() { - @Override - public void subscribe(Subscriber s) { - s.onSubscribe(new BooleanSubscription()); - s.onNext(Completable.never()); - s.onNext(Completable.never()); - s.onNext(Completable.never()); - s.onNext(Completable.never()); - s.onComplete(); - } + Flowable.fromPublisher(s -> { + s.onSubscribe(new BooleanSubscription()); + s.onNext(Completable.never()); + s.onNext(Completable.never()); + s.onNext(Completable.never()); + s.onNext(Completable.never()); + s.onComplete(); }), 1 ) .test() @@ -84,29 +81,14 @@ public void errorRace() { final PublishProcessor pp1 = PublishProcessor.create(); final PublishProcessor pp2 = PublishProcessor.create(); - TestObserver to = Completable.concat(pp1.map(new Function() { - @Override - public Completable apply(Integer v) throws Exception { - return pp2.ignoreElements(); - } - })).test(); + TestObserver to = Completable.concat(pp1.map(_ -> pp2.ignoreElements())).test(); pp1.onNext(1); final TestException ex = new TestException(); - Runnable r1 = new Runnable() { - @Override - public void run() { - pp1.onError(ex); - } - }; - Runnable r2 = new Runnable() { - @Override - public void run() { - pp2.onError(ex); - } - }; + Runnable r1 = () -> pp1.onError(ex); + Runnable r2 = () -> pp2.onError(ex); TestHelper.race(r1, r2); @@ -123,11 +105,8 @@ public void run() { @Test public void synchronousFusedCrash() { - Completable.concat(Flowable.range(1, 2).map(new Function() { - @Override - public Completable apply(Integer v) throws Exception { - throw new TestException(); - } + Completable.concat(Flowable.range(1, 2).map(_ -> { + throw new TestException(); })) .test() .assertFailure(TestException.class); @@ -169,7 +148,7 @@ public void arrayCancelled() { public void arrayFirstCancels() { final TestObserver to = new TestObserver<>(); - Completable.concatArray(new Completable() { + Completable.concatArray(new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(Disposable.empty()); @@ -193,7 +172,7 @@ public void iterableCancelled() { public void iterableFirstCancels() { final TestObserver to = new TestObserver<>(); - Completable.concat(Arrays.asList(new Completable() { + Completable.concat(Arrays.asList(new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(Disposable.empty()); @@ -217,19 +196,9 @@ public void arrayCancelRace() { final TestObserver to = new TestObserver<>(); - Runnable r1 = new Runnable() { - @Override - public void run() { - c.subscribe(to); - } - }; + Runnable r1 = () -> c.subscribe(to); - Runnable r2 = new Runnable() { - @Override - public void run() { - to.dispose(); - } - }; + Runnable r2 = () -> to.dispose(); TestHelper.race(r1, r2); } @@ -246,19 +215,9 @@ public void iterableCancelRace() { final TestObserver to = new TestObserver<>(); - Runnable r1 = new Runnable() { - @Override - public void run() { - c.subscribe(to); - } - }; + Runnable r1 = () -> c.subscribe(to); - Runnable r2 = new Runnable() { - @Override - public void run() { - to.dispose(); - } - }; + Runnable r2 = () -> to.dispose(); TestHelper.race(r1, r2); } @@ -272,15 +231,12 @@ public void noInterrupt() throws InterruptedException { final boolean[] interrupted = { false }; for (int i = 0; i < count; i++) { - Completable c0 = Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - try { - Thread.sleep(30); - } catch (InterruptedException e) { - System.out.println("Interrupted! " + Thread.currentThread()); - interrupted[0] = true; - } + Completable c0 = Completable.fromAction(() -> { + try { + Thread.sleep(30); + } catch (InterruptedException e) { + System.out.println("Interrupted! " + Thread.currentThread()); + interrupted[0] = true; } }); Completable.concat(Arrays.asList(Completable.complete() @@ -288,12 +244,7 @@ public void run() throws Exception { .observeOn(Schedulers.cached()), c0) ) - .subscribe(new Action() { - @Override - public void run() throws Exception { - latch.countDown(); - } - }); + .subscribe(() -> latch.countDown()); } latch.await(); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCreateTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCreateTest.java index 04c69a3c70..da71274c6c 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCreateTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableCreateTest.java @@ -23,7 +23,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.Cancellable; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.testsupport.TestHelper; @@ -35,15 +34,12 @@ public void basic() { try { final Disposable d = Disposable.empty(); - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - e.setDisposable(d); + Completable.create(e -> { + e.setDisposable(d); - e.onComplete(); - e.onError(new TestException()); - e.onComplete(); - } + e.onComplete(); + e.onError(new TestException()); + e.onComplete(); }) .test() .assertResult(); @@ -63,21 +59,13 @@ public void basicWithCancellable() { final Disposable d1 = Disposable.empty(); final Disposable d2 = Disposable.empty(); - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - e.setDisposable(d1); - e.setCancellable(new Cancellable() { - @Override - public void cancel() throws Exception { - d2.dispose(); - } - }); - - e.onComplete(); - e.onError(new TestException()); - e.onComplete(); - } + Completable.create(e -> { + e.setDisposable(d1); + e.setCancellable(() -> d2.dispose()); + + e.onComplete(); + e.onError(new TestException()); + e.onComplete(); }) .test() .assertResult(); @@ -97,15 +85,12 @@ public void basicWithError() { try { final Disposable d = Disposable.empty(); - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - e.setDisposable(d); + Completable.create(e -> { + e.setDisposable(d); - e.onError(new TestException()); - e.onComplete(); - e.onError(new TestException("second")); - } + e.onError(new TestException()); + e.onComplete(); + e.onError(new TestException("second")); }) .test() .assertFailure(TestException.class); @@ -120,11 +105,8 @@ public void subscribe(CompletableEmitter e) throws Exception { @Test public void callbackThrows() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - throw new TestException(); - } + Completable.create(_ -> { + throw new TestException(); }) .test() .assertFailure(TestException.class); @@ -132,45 +114,32 @@ public void subscribe(CompletableEmitter e) throws Exception { @Test public void onErrorNull() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - e.onError(null); - } - }) + Completable.create(e -> e.onError(null)) .test() .assertFailure(NullPointerException.class); } @Test public void dispose() { - TestHelper.checkDisposed(Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - e.onComplete(); - } - })); + TestHelper.checkDisposed(Completable.create(CompletableEmitter::onComplete)); } @Test public void onErrorThrows() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - Disposable d = Disposable.empty(); - e.setDisposable(d); - - try { - e.onError(new IOException()); - fail("Should have thrown"); - } catch (TestException ex) { - // expected - } - - assertTrue(d.isDisposed()); - assertTrue(e.isDisposed()); + Completable.create(e -> { + Disposable d = Disposable.empty(); + e.setDisposable(d); + + try { + e.onError(new IOException()); + fail("Should have thrown"); + } catch (TestException ex) { + // expected } - }).subscribe(new CompletableObserver() { + + assertTrue(d.isDisposed()); + assertTrue(e.isDisposed()); + }).subscribe(new CompletableObserver() /* NFI */ { @Override public void onSubscribe(Disposable d) { @@ -191,23 +160,20 @@ public void onComplete() { @Test public void onCompleteThrows() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - Disposable d = Disposable.empty(); - e.setDisposable(d); - - try { - e.onComplete(); - fail("Should have thrown"); - } catch (TestException ex) { - // expected - } + Completable.create(e -> { + Disposable d = Disposable.empty(); + e.setDisposable(d); - assertTrue(d.isDisposed()); - assertTrue(e.isDisposed()); + try { + e.onComplete(); + fail("Should have thrown"); + } catch (TestException ex) { + // expected } - }).subscribe(new CompletableObserver() { + + assertTrue(d.isDisposed()); + assertTrue(e.isDisposed()); + }).subscribe(new CompletableObserver() /* NFI */ { @Override public void onSubscribe(Disposable d) { @@ -228,19 +194,16 @@ public void onComplete() { @Test public void onErrorThrows2() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - try { - e.onError(new IOException()); - fail("Should have thrown"); - } catch (TestException ex) { - // expected - } - - assertTrue(e.isDisposed()); + Completable.create(e -> { + try { + e.onError(new IOException()); + fail("Should have thrown"); + } catch (TestException ex) { + // expected } - }).subscribe(new CompletableObserver() { + + assertTrue(e.isDisposed()); + }).subscribe(new CompletableObserver() /* NFI */ { @Override public void onSubscribe(Disposable d) { @@ -261,19 +224,16 @@ public void onComplete() { @Test public void onCompleteThrows2() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - try { - e.onComplete(); - fail("Should have thrown"); - } catch (TestException ex) { - // expected - } - - assertTrue(e.isDisposed()); + Completable.create(e -> { + try { + e.onComplete(); + fail("Should have thrown"); + } catch (TestException ex) { + // expected } - }).subscribe(new CompletableObserver() { + + assertTrue(e.isDisposed()); + }).subscribe(new CompletableObserver() /* NFI */ { @Override public void onSubscribe(Disposable d) { @@ -297,12 +257,9 @@ public void tryOnError() { List errors = TestHelper.trackPluginErrors(); try { final Boolean[] response = { null }; - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter e) throws Exception { - e.onComplete(); - response[0] = e.tryOnError(new TestException()); - } + Completable.create(e -> { + e.onComplete(); + response[0] = e.tryOnError(new TestException()); }) .test() .assertResult(); @@ -317,11 +274,9 @@ public void subscribe(CompletableEmitter e) throws Exception { @Test public void emitterHasToString() { - Completable.create(new CompletableOnSubscribe() { - @Override - public void subscribe(CompletableEmitter emitter) throws Exception { - assertTrue(emitter.toString().contains(CompletableCreate.Emitter.class.getSimpleName())); - } - }).test().assertEmpty(); + Completable.create(emitter -> + assertTrue(emitter.toString().contains(CompletableCreate.Emitter.class.getSimpleName()))) + .test() + .assertEmpty(); } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelaySubscriptionTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelaySubscriptionTest.java index 591590a9d0..4668bd8677 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelaySubscriptionTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelaySubscriptionTest.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.Action; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.schedulers.TestScheduler; import io.reactivex.rxjava4.subjects.CompletableSubject; @@ -33,12 +32,7 @@ public class CompletableDelaySubscriptionTest extends RxJavaTest { public void normal() { final AtomicInteger counter = new AtomicInteger(); - Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - counter.incrementAndGet(); - } - }) + Completable.fromAction(() -> counter.incrementAndGet()) .delaySubscription(100, TimeUnit.MILLISECONDS) .test() .awaitDone(5, TimeUnit.SECONDS) @@ -51,13 +45,10 @@ public void run() throws Exception { public void error() { final AtomicInteger counter = new AtomicInteger(); - Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - counter.incrementAndGet(); + Completable.fromAction(() -> { + counter.incrementAndGet(); - throw new TestException(); - } + throw new TestException(); }) .delaySubscription(100, TimeUnit.MILLISECONDS) .test() @@ -73,12 +64,7 @@ public void disposeBeforeTime() { final AtomicInteger counter = new AtomicInteger(); - Completable result = Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - counter.incrementAndGet(); - } - }) + Completable result = Completable.fromAction(() -> counter.incrementAndGet()) .delaySubscription(100, TimeUnit.MILLISECONDS, scheduler); TestObserver to = result.test(); @@ -100,12 +86,7 @@ public void timestep() { TestScheduler scheduler = new TestScheduler(); final AtomicInteger counter = new AtomicInteger(); - Completable result = Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - counter.incrementAndGet(); - } - }) + Completable result = Completable.fromAction(() -> counter.incrementAndGet()) .delaySubscription(100, TimeUnit.MILLISECONDS, scheduler); TestObserver to = result.test(); @@ -123,13 +104,10 @@ public void timestepError() { TestScheduler scheduler = new TestScheduler(); final AtomicInteger counter = new AtomicInteger(); - Completable result = Completable.fromAction(new Action() { - @Override - public void run() throws Exception { - counter.incrementAndGet(); + Completable result = Completable.fromAction(() -> { + counter.incrementAndGet(); - throw new TestException(); - } + throw new TestException(); }) .delaySubscription(100, TimeUnit.MILLISECONDS, scheduler); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelayTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelayTest.java index 79d1ec6808..296a1d4394 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelayTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDelayTest.java @@ -45,12 +45,9 @@ public void onErrorCalledOnScheduler() throws Exception { Completable.error(new Exception()) .delay(0, TimeUnit.MILLISECONDS, Schedulers.newThread()) - .doOnError(new Consumer() { - @Override - public void accept(Throwable throwable) throws Exception { - thread.set(Thread.currentThread()); - latch.countDown(); - } + .doOnError(_ -> { + thread.set(Thread.currentThread()); + latch.countDown(); }) .onErrorComplete() .subscribe(); @@ -67,12 +64,7 @@ public void disposed() { @Test public void doubleOnSubscribe() { - TestHelper.checkDoubleOnSubscribeCompletable(new Function() { - @Override - public CompletableSource apply(Completable c) throws Exception { - return c.delay(1, TimeUnit.MINUTES); - } - }); + TestHelper.checkDoubleOnSubscribeCompletable(c -> c.delay(1, TimeUnit.MINUTES)); } @Test diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDetachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDetachTest.java index 0bf3b80b89..f3c882bf3c 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDetachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDetachTest.java @@ -23,7 +23,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.processors.PublishProcessor; import io.reactivex.rxjava4.testsupport.TestHelper; @@ -33,12 +32,7 @@ public class CompletableDetachTest extends RxJavaTest { @Test public void doubleSubscribe() { - TestHelper.checkDoubleOnSubscribeCompletable(new Function() { - @Override - public CompletableSource apply(Completable m) throws Exception { - return m.onTerminateDetach(); - } - }); + TestHelper.checkDoubleOnSubscribeCompletable(Completable::onTerminateDetach); } @Test @@ -67,7 +61,7 @@ public void cancelDetaches() throws Exception { Disposable d = Disposable.empty(); final WeakReference wr = new WeakReference<>(d); - TestObserver to = new Completable() { + TestObserver to = new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(wr.get()); @@ -93,7 +87,7 @@ public void completeDetaches() throws Exception { Disposable d = Disposable.empty(); final WeakReference wr = new WeakReference<>(d); - TestObserver to = new Completable() { + TestObserver to = new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(wr.get()); @@ -119,7 +113,7 @@ public void errorDetaches() throws Exception { Disposable d = Disposable.empty(); final WeakReference wr = new WeakReference<>(d); - TestObserver to = new Completable() { + TestObserver to = new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(wr.get()); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDisposeOnTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDisposeOnTest.java index ac1f62f863..e0c42a4b5f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDisposeOnTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDisposeOnTest.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.Action; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; @@ -102,12 +101,7 @@ public void normal() { final int[] call = { 0 }; Completable.complete() - .doOnDispose(new Action() { - @Override - public void run() throws Exception { - call[0]++; - } - }) + .doOnDispose(() -> call[0]++) .unsubscribeOn(scheduler) .test() .assertResult(); @@ -124,12 +118,7 @@ public void error() { final int[] call = { 0 }; Completable.error(new TestException()) - .doOnDispose(new Action() { - @Override - public void run() throws Exception { - call[0]++; - } - }) + .doOnDispose(() -> call[0]++) .unsubscribeOn(scheduler) .test() .assertFailure(TestException.class); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoFinallyTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoFinallyTest.java index b8a3b0f179..2f027abe32 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoFinallyTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoFinallyTest.java @@ -57,12 +57,8 @@ public void normalError() { @Test public void doubleOnSubscribe() { - TestHelper.checkDoubleOnSubscribeCompletable(new Function() { - @Override - public Completable apply(Completable f) throws Exception { - return f.doFinally(CompletableDoFinallyTest.this); - } - }); + TestHelper.checkDoubleOnSubscribeCompletable( + (Function) f -> f.doFinally(CompletableDoFinallyTest.this)); } @Test @@ -70,11 +66,8 @@ public void actionThrows() { List errors = TestHelper.trackPluginErrors(); try { Completable.complete() - .doFinally(new Action() { - @Override - public void run() throws Exception { - throw new TestException(); - } + .doFinally(() -> { + throw new TestException(); }) .test() .assertResult() diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnLifecycleTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnLifecycleTest.java index 33a6cfb1ad..ff289d0f75 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnLifecycleTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnLifecycleTest.java @@ -70,7 +70,7 @@ public void onSubscribeCrash() throws Throwable { Disposable bs = Disposable.empty(); - new Completable() { + new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(bs); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnTest.java index 82cd1473d9..60529e84f6 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDoOnTest.java @@ -31,11 +31,8 @@ public class CompletableDoOnTest extends RxJavaTest { @Test public void successAcceptThrows() { - Completable.complete().doOnEvent(new Consumer() { - @Override - public void accept(Throwable e) throws Exception { - throw new TestException(); - } + Completable.complete().doOnEvent(_ -> { + throw new TestException(); }) .test() .assertFailure(TestException.class); @@ -43,11 +40,9 @@ public void accept(Throwable e) throws Exception { @Test public void errorAcceptThrows() { - TestObserverEx to = Completable.error(new TestException("Outer")).doOnEvent(new Consumer() { - @Override - public void accept(Throwable e) throws Exception { - throw new TestException("Inner"); - } + TestObserverEx to = Completable.error(new TestException("Outer")) + .doOnEvent(_ -> { + throw new TestException("Inner"); }) .to(TestHelper.testConsumer()) .assertFailure(CompositeException.class); @@ -65,12 +60,7 @@ public void doOnDisposeCalled() { assertFalse(atomicBoolean.get()); Completable.complete() - .doOnDispose(new Action() { - @Override - public void run() throws Exception { - atomicBoolean.set(true); - } - }) + .doOnDispose(() -> atomicBoolean.set(true)) .test() .assertResult() .dispose(); @@ -84,7 +74,7 @@ public void onSubscribeCrash() { try { final Disposable bs = Disposable.empty(); - new Completable() { + new Completable() /* NFI */ { @Override protected void subscribeActual(CompletableObserver observer) { observer.onSubscribe(bs); @@ -92,11 +82,8 @@ protected void subscribeActual(CompletableObserver observer) { observer.onComplete(); } } - .doOnSubscribe(new Consumer() { - @Override - public void accept(Disposable d) throws Exception { - throw new TestException("First"); - } + .doOnSubscribe(_ -> { + throw new TestException("First"); }) .to(TestHelper.testConsumer()) .assertFailureAndMessage(TestException.class, "First"); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java index 1db39e2c9e..7c43fe7fae 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java @@ -21,11 +21,11 @@ import java.time.Duration; import java.util.*; import java.util.concurrent.*; +import java.util.concurrent.Flow.*; import java.util.concurrent.atomic.*; import org.junit.Test; import org.mockito.Mockito; -import static java.util.concurrent.Flow.*; import com.google.common.base.Ticker; import com.google.common.cache.*; @@ -2013,7 +2013,7 @@ public void groupByEvictionCancellationOfSource5933() { @Override public Map apply(final Consumer action) throws Exception { return CacheBuilder.newBuilder() // - .expireAfterAccess(5, TimeUnit.SECONDS).removalListener(new RemovalListener() { + .expireAfterAccess(Duration.ofSeconds(5)).removalListener(new RemovalListener() { @Override public void onRemoval(RemovalNotification notification) { try {