4.x: Cleanup - Migration; lambdas expressions (#8188)

* 4.x: Cleanup - Migration; lambdas expressions

* Fix message format error accidentally introduced in test

Author: akarnokd

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

@@ -238,9 +238,7 @@ default Flowable<T> toFlowable() {
     default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
         Objects.requireNonNull(executor, "executir is null");
         var me = this;
-        return Flowable.virtualCreate(emitter -> {
-            me.forEach(emitter::emit).await(emitter.canceller());
-        }, executor);
+        return Flowable.virtualCreate(emitter -> me.forEach(emitter::emit).await(emitter.canceller()), executor);
     }
 
     /**
@@ -269,13 +267,11 @@ default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
         Objects.requireNonNull(transformer, "transformer is null");
         Objects.requireNonNull(executor, "executor is null");
         var me = this;
-        return create(emitter -> {
-            me.forEach((item, stopper) -> {
-                // System.out.println("item " + item);
-                transformer.transform(item, emitter, stopper);
-            }, emitter.canceller(), executor)
-            .await(emitter.canceller());
-        }, executor);
+        return create(emitter -> me.forEach((item, stopper) -> {
+            // System.out.println("item " + item);
+            transformer.transform(item, emitter, stopper);
+        }, emitter.canceller(), executor)
+        .await(emitter.canceller()), executor);
     }
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo

src/main/java/io/reactivex/rxjava4/internal/operators/parallel/ParallelJoin.java

@@ -216,7 +216,6 @@ void drainLoop() {
             int missed = 1;
 
             JoinInnerSubscriber<T>[] s = this.subscribers;
-            int n = s.length;
             Subscriber<? super T> a = this.downstream;
 
             for (;;) {
@@ -387,7 +386,6 @@ void drainLoop() {
             int missed = 1;
 
             JoinInnerSubscriber<T>[] s = this.subscribers;
-            int n = s.length;
             Subscriber<? super T> a = this.downstream;
 
             for (;;) {

src/main/java/io/reactivex/rxjava4/internal/util/AwaitCoordinatorStatic.java

@@ -122,9 +122,7 @@ static <T> T awaitFirst(Iterable<? extends CompletionStage<? extends T>> stages,
         var winner = new CompletableFuture<CompletionStage<? extends T>>();
 
         for (var stage : stages) {
-            stage.whenComplete((_, _) -> {
-                winner.complete(stage);
-            });
+            stage.whenComplete((_, _) -> winner.complete(stage));
         }
         return await(winner, canceller).toCompletableFuture().getNow(null);
     }
@@ -144,9 +142,7 @@ static <T> CompletionStage<T> awaitFirstStage(
         var winner = new CompletableFuture<CompletionStage<? extends T>>();
 
         for (var stage : stages) {
-            stage.whenComplete((_, _) -> {
-                winner.complete(stage);
-            });
+            stage.whenComplete((_, _) -> winner.complete(stage));
         }
         return (CompletionStage<T>)await(winner, canceller).toCompletableFuture().getNow(null);
     }
@@ -167,9 +163,7 @@ static <T> int awaitFirstIndex(
         for (int i = 0; i < stages.size(); i++) {
             var stage = stages.get(i);
             var fi = i;
-            stage.whenComplete((_, _) -> {
-                winner.complete(fi);
-            });
+            stage.whenComplete((_, _) -> winner.complete(fi));
         }
         return await(winner, canceller);
     }

src/test/java/io/reactivex/rxjava4/internal/observers/FutureSingleObserverTest.java

@@ -47,9 +47,7 @@ public void cancel() {
             fail("Should have thrown!");
         } catch (CancellationException ex) {
             // expected
-        } catch (InterruptedException ex) {
-            throw new AssertionError(ex);
-        } catch (ExecutionException ex) {
+        } catch (InterruptedException | ExecutionException ex) {
             throw new AssertionError(ex);
         }
 
@@ -58,11 +56,7 @@ public void cancel() {
             fail("Should have thrown!");
         } catch (CancellationException ex) {
             // expected
-        } catch (InterruptedException ex) {
-            throw new AssertionError(ex);
-        } catch (ExecutionException ex) {
-            throw new AssertionError(ex);
-        } catch (TimeoutException ex) {
+        } catch (InterruptedException | ExecutionException | TimeoutException ex) {
             throw new AssertionError(ex);
         }
     }

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java

@@ -1736,11 +1736,11 @@ public void delayErrorCompleteMoreWorkInGroup() {
     public void groupSyncFusionRejected() {
         Flowable.just(1)
         .groupBy(_ -> 1)
-        .doOnNext(g -> {
-            g.subscribeWith(new TestSubscriberEx<Integer>().setInitialFusionMode(QueueFuseable.SYNC))
-            .assertFuseable()
-            .assertFusionMode(QueueFuseable.NONE);
-        })
+        .doOnNext(g ->
+                g.subscribeWith(new TestSubscriberEx<Integer>()
+                                .setInitialFusionMode(QueueFuseable.SYNC))
+        .assertFuseable()
+        .assertFusionMode(QueueFuseable.NONE))
         .test()
         .assertComplete();
     }
@@ -1755,11 +1755,7 @@ public void subscribeAbandonRace() throws Throwable {
             CountDownLatch cdl = new CountDownLatch(1);
 
             pp.groupBy(_ -> 1)
-            .doOnNext(g -> {
-                TestHelper.raceOther(() -> {
-                    g.subscribe(ts);
-                }, cdl);
-            })
+            .doOnNext(g -> TestHelper.raceOther(() -> g.subscribe(ts), cdl))
             .test();
 
             pp.onNext(1);
@@ -1895,9 +1891,7 @@ public void issue6974Part2Case1ObserveOn() {
         Flowable
         .range(1, 500_000)
         .map(i -> i % groups)
-        .doOnCancel(() -> {
-            System.out.println("Cancelling upstream");
-        })
+        .doOnCancel(() -> System.out.println("Cancelling upstream"))
         .groupBy(i -> i, i -> i, false, groupByBufferSize,
                               sizeCap(groups * 2, notifyOnExplicitEviction))
         .flatMap(gf -> gf
@@ -1946,9 +1940,7 @@ public void issue6974Part2Case1ObserveOnNoCap() {
         Flowable
         .range(1, 500_000)
         .map(i -> i % groups)
-        .doOnRequest(v -> {
-            System.out.println("Source: " + v);
-        })
+        .doOnRequest(v -> System.out.println("Source: " + v))
         .groupBy(i -> i)
         .flatMap(gf -> gf
                      .observeOn(Schedulers.computation())
@@ -1971,9 +1963,7 @@ public void issue6974Part2Case1ObserveOnNoCapHide() {
         Flowable
         .range(1, 500_000)
         .map(i -> i % groups)
-        .doOnRequest(v -> {
-            System.out.println("Source: " + v);
-        })
+        .doOnRequest(v -> System.out.println("Source: " + v))
         .groupBy(i -> i)
         .flatMap(gf -> gf
                      .hide()

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupJoinTest.java

@@ -551,9 +551,7 @@ public void disposeAfterOnNext() {
         TestSubscriber<Integer> ts = new TestSubscriber<>();
 
         pp1.groupJoin(pp2, _ -> Flowable.never(), _ -> Flowable.never(), (a, _) -> a)
-        .doOnNext(_ -> {
-            ts.cancel();
-        })
+        .doOnNext(_ -> ts.cancel())
         .subscribe(ts);
 
         pp2.onNext(1);

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableSwitchTest.java

@@ -947,9 +947,7 @@ public void fusedBoundary() {
         Flowable.range(1, 10000)
         .switchMap((Function<Integer, Flowable<Object>>) _ -> Flowable.just(2).hide()
         .observeOn(Schedulers.single())
-        .map((Function<Integer, Object>) _ -> {
-            return Thread.currentThread().getName();
-        }))
+        .map((Function<Integer, Object>) _ -> Thread.currentThread().getName()))
         .to(TestHelper.<Object>testConsumer())
         .awaitDone(5, TimeUnit.SECONDS)
         .assertNever(thread)
@@ -1110,9 +1108,7 @@ public void cancellationShouldTriggerInnerCancellationRace() throws Throwable {
 
         int n = 10_000;
         for (int i = 0; i < n; i++) {
-            Flowable.<Integer>create(it -> {
-                it.onNext(0);
-            }, BackpressureStrategy.MISSING)
+            Flowable.<Integer>create(it -> it.onNext(0), BackpressureStrategy.MISSING)
             .switchMap(_ -> createFlowable(inner))
             .observeOn(Schedulers.computation())
             .doFinally(outer::incrementAndGet)
@@ -1129,12 +1125,8 @@ Flowable<Integer> createFlowable(AtomicInteger inner) {
         return Flowable.<Integer>unsafeCreate(s -> {
             SerializedSubscriber<Integer> it = new SerializedSubscriber<>(s);
             it.onSubscribe(new BooleanSubscription());
-            Schedulers.cached().scheduleDirect(() -> {
-                it.onNext(1);
-            }, 0, TimeUnit.MILLISECONDS);
-            Schedulers.cached().scheduleDirect(() -> {
-                it.onNext(2);
-            }, 0, TimeUnit.MILLISECONDS);
+            Schedulers.cached().scheduleDirect(() -> it.onNext(1), 0, TimeUnit.MILLISECONDS);
+            Schedulers.cached().scheduleDirect(() -> it.onNext(2), 0, TimeUnit.MILLISECONDS);
         })
         .doFinally(inner::incrementAndGet);
     }

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableToListTest.java

@@ -174,9 +174,7 @@ public void listWithBlockingFirst() {
     static void await(CyclicBarrier cb) {
         try {
             cb.await();
-        } catch (InterruptedException ex) {
-            ex.printStackTrace();
-        } catch (BrokenBarrierException ex) {
+        } catch (InterruptedException | BrokenBarrierException ex) {
             ex.printStackTrace();
         }
     }

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableToSortedListTest.java

@@ -145,9 +145,7 @@ public void withFollowingFirst() {
     static void await(CyclicBarrier cb) {
         try {
             cb.await();
-        } catch (InterruptedException ex) {
-            ex.printStackTrace();
-        } catch (BrokenBarrierException ex) {
+        } catch (InterruptedException | BrokenBarrierException ex) {
             ex.printStackTrace();
         }
     }

src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCreateTest.java

@@ -586,9 +586,7 @@ public void serializedEnqueueAndDrainRace() throws Throwable {
             })
             .doOnNext(v -> {
                 if (v == 1) {
-                    TestHelper.raceOther(() -> {
-                        ref.get().onNext(2);
-                    }, cdl);
+                    TestHelper.raceOther(() -> ref.get().onNext(2), cdl);
                     ref.get().onNext(3);
                 }
             })