Skip to content

4.x: Cleanup + lambda instead of inner class check #8179

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 3 commits into from
Jun 25, 2026
Merged

4.x: Cleanup + lambda instead of inner class check #8179

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c0a8511
4.x: Cleanup + lambda instead of inner class check
akarnokd Jun 25, 2026
e2b8e82
Merge branch '4.x' into 4xVerifyLambdaUsageTest
akarnokd Jun 25, 2026
d58230c
Fix merge main error
akarnokd Jun 25, 2026
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
2 changes: 1 addition & 1 deletion ...reactivex/rxjava4/core/SchedulerTest.java → ...tivex/rxjava4/core/SchedulerMainTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@

import io.reactivex.rxjava4.schedulers.Schedulers;

public class SchedulerTest {
public class SchedulerMainTest {
private static final String DRIFT_USE_NANOTIME = "rxjava4.scheduler.use-nanotime";

@After
Expand Down
34 changes: 14 additions & 20 deletions src/test/java/io/reactivex/rxjava4/disposables/CompositeDisposableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,17 +56,14 @@ public void shouldUnsubscribeAll() throws InterruptedException {

final List<Thread> threads = new ArrayList<>();
for (int i = 0; i < count; i++) {
final Thread t = new Thread() /* NFI */ {
@Override
public void run() {
try {
start.await();
cd.dispose();
} catch (final InterruptedException e) {
fail(e.getMessage());
}
final Thread t = new Thread(() -> {
try {
start.await();
cd.dispose();
} catch (final InterruptedException e) {
fail(e.getMessage());
}
};
});
t.start();
threads.add(t);
}
Expand Down Expand Up @@ -201,17 +198,14 @@ public void unsubscribeIdempotenceConcurrently()

final List<Thread> threads = new ArrayList<>();
for (int i = 0; i < count; i++) {
final Thread t = new Thread() /* NFI */ {
@Override
public void run() {
try {
start.await();
cd.dispose();
} catch (final InterruptedException e) {
fail(e.getMessage());
}
final Thread t = new Thread(() -> {
try {
start.await();
cd.dispose();
} catch (final InterruptedException e) {
fail(e.getMessage());
}
};
});
t.start();
threads.add(t);
}
Expand Down
42 changes: 18 additions & 24 deletions src/test/java/io/reactivex/rxjava4/disposables/SequentialDisposableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,19 +131,16 @@ public void settingUnderlyingWhenUnsubscribedCausesImmediateUnsubscriptionConcur

final List<Thread> threads = new ArrayList<>();
for (int i = 0; i < count; i++) {
final Thread t = new Thread() /* NFI */ {
@Override
public void run() {
try {
start.await();
serialDisposable.dispose();
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
final Thread t = new Thread(() -> {
try {
start.await();
serialDisposable.dispose();
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
};
});
t.start();
threads.add(t);
}
Expand Down Expand Up @@ -174,19 +171,16 @@ public void concurrentSetDisposableShouldNotInterleave()
final Disposable subscription = mock(Disposable.class);
subscriptions.add(subscription);

final Thread t = new Thread() /* NFI */ {
@Override
public void run() {
try {
start.await();
serialDisposable.update(subscription);
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
final Thread t = new Thread(() -> {
try {
start.await();
serialDisposable.update(subscription);
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
};
});
t.start();
threads.add(t);
}
Expand Down
42 changes: 18 additions & 24 deletions src/test/java/io/reactivex/rxjava4/disposables/SerialDisposableTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,19 +131,16 @@ public void settingUnderlyingWhenUnsubscribedCausesImmediateUnsubscriptionConcur

final List<Thread> threads = new ArrayList<>();
for (int i = 0; i < count; i++) {
final Thread t = new Thread() /* NFI */ {
@Override
public void run() {
try {
start.await();
serialDisposable.dispose();
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
final Thread t = new Thread(() -> {
try {
start.await();
serialDisposable.dispose();
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
};
});
t.start();
threads.add(t);
}
Expand Down Expand Up @@ -174,19 +171,16 @@ public void concurrentSetDisposableShouldNotInterleave()
final Disposable subscription = mock(Disposable.class);
subscriptions.add(subscription);

final Thread t = new Thread() /* NFI */ {
@Override
public void run() {
try {
start.await();
serialDisposable.set(subscription);
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
final Thread t = new Thread(() -> {
try {
start.await();
serialDisposable.set(subscription);
} catch (InterruptedException e) {
fail(e.getMessage());
} finally {
end.countDown();
}
};
});
t.start();
threads.add(t);
}
Expand Down
24 changes: 9 additions & 15 deletions src/test/java/io/reactivex/rxjava4/internal/operators/flowable/BlockingFlowableNextTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,31 +34,25 @@
public class BlockingFlowableNextTest extends RxJavaTest {

private void fireOnNextInNewThread(final FlowableProcessor<String> o, final String value) {
new Thread() /* NFI */ {
@Override
public void run() {
new Thread(() -> {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
// ignore
}
o.onNext(value);
}
}.start();
}).start();
}

private void fireOnErrorInNewThread(final FlowableProcessor<String> o) {
new Thread() /* NFI */ {
@Override
public void run() {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
// ignore
}
o.onError(new TestException());
new Thread(() -> {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
// ignore
}
}.start();
o.onError(new TestException());
}).start();
}

@Test
Expand Down
2 changes: 1 addition & 1 deletion src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCountTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -528,7 +528,7 @@ public void disposed() {
@Test
public void noOpConnect() {
final int[] calls = { 0 };
Flowable<Integer> f = new ConnectableFlowable<Integer>() {
Flowable<Integer> f = new ConnectableFlowable<Integer>() /* NFI */ {
@Override
public void connect(Consumer<? super Disposable> connection) {
calls[0]++;
Expand Down
16 changes: 8 additions & 8 deletions ...ava/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayEagerTruncateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -664,7 +664,7 @@ public boolean isDisposed() {

@Test
public void boundedReplayBuffer() {
BoundedReplayBuffer<Integer> buf = new BoundedReplayBuffer<Integer>(true) {
BoundedReplayBuffer<Integer> buf = new BoundedReplayBuffer<Integer>(true) /* NFI */ {
private static final long serialVersionUID = -9081211580719235896L;

@Override
Expand Down Expand Up @@ -995,7 +995,7 @@ public void unsafeChildThrows() {
.doOnNext(_ -> count.getAndIncrement())
.replay().autoConnect();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>() /* NFI */ {
@Override
public void onNext(Integer t) {
throw new TestException();
Expand Down Expand Up @@ -1257,7 +1257,7 @@ public void connectConsumerThrows() {
public void badSource() {
List<Throwable> errors = TestHelper.trackPluginErrors();
try {
new Flowable<Integer>() {
new Flowable<Integer>() /* NFI */ {
@Override
protected void subscribeActual(Subscriber<? super Integer> subscriber) {
subscriber.onSubscribe(new BooleanSubscription());
Expand Down Expand Up @@ -1342,7 +1342,7 @@ public void unsubscribeReplayRace() {
public void reentrantOnNext() {
final PublishProcessor<Integer> pp = PublishProcessor.create();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>() /* NFI */ {
@Override
public void onNext(Integer t) {
if (t == 1) {
Expand All @@ -1364,7 +1364,7 @@ public void onNext(Integer t) {
public void reentrantOnNextBound() {
final PublishProcessor<Integer> pp = PublishProcessor.create();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>() /* NFI */ {
@Override
public void onNext(Integer t) {
if (t == 1) {
Expand All @@ -1386,7 +1386,7 @@ public void onNext(Integer t) {
public void reentrantOnNextCancel() {
final PublishProcessor<Integer> pp = PublishProcessor.create();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>() /* NFI */ {
@Override
public void onNext(Integer t) {
if (t == 1) {
Expand All @@ -1408,7 +1408,7 @@ public void onNext(Integer t) {
public void reentrantOnNextCancelBounded() {
final PublishProcessor<Integer> pp = PublishProcessor.create();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>() /* NFI */ {
@Override
public void onNext(Integer t) {
if (t == 1) {
Expand Down Expand Up @@ -1517,7 +1517,7 @@ public void sizedTruncation() {
public void delayedUpstreamOnSubscribe() {
final Subscriber<?>[] sub = { null };

new Flowable<Integer>() {
new Flowable<Integer>() /* NFI */ {
@Override
protected void subscribeActual(Subscriber<? super Integer> s) {
sub[0] = s;
Expand Down
Loading