3.x: Fix Schedulers.from to honor interruptibleWorker across methods (#7203)

akarnokd · web-flow · commit 26dddf51efed · 2021-03-05T10:30:00.000+01:00
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/AbstractDirectTask.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/AbstractDirectTask.java
@@ -32,14 +32,17 @@ abstract class AbstractDirectTask
 
     protected final Runnable runnable;
 
+    protected final boolean interruptOnCancel;
+
     protected Thread runner;
 
     protected static final FutureTask<Void> FINISHED = new FutureTask<>(Functions.EMPTY_RUNNABLE, null);
 
     protected static final FutureTask<Void> DISPOSED = new FutureTask<>(Functions.EMPTY_RUNNABLE, null);
 
-    AbstractDirectTask(Runnable runnable) {
+    AbstractDirectTask(Runnable runnable, boolean interruptOnCancel) {
         this.runnable = runnable;
+        this.interruptOnCancel = interruptOnCancel;
     }
 
     @Override
@@ -48,7 +51,7 @@ public final void dispose() {
         if (f != FINISHED && f != DISPOSED) {
             if (compareAndSet(f, DISPOSED)) {
                 if (f != null) {
-                    f.cancel(runner != Thread.currentThread());
+                    cancelFuture(f);
                 }
             }
         }
@@ -67,7 +70,7 @@ public final void setFuture(Future<?> future) {
                 break;
             }
             if (f == DISPOSED) {
-                future.cancel(runner != Thread.currentThread());
+                cancelFuture(future);
                 break;
             }
             if (compareAndSet(f, future)) {
@@ -76,6 +79,14 @@ public final void setFuture(Future<?> future) {
         }
     }
 
+    private void cancelFuture(Future<?> future) {
+        if (runner == Thread.currentThread()) {
+            future.cancel(false);
+        } else {
+            future.cancel(interruptOnCancel);
+        }
+    }
+
     @Override
     public Runnable getWrappedRunnable() {
         return runnable;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/ExecutorScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/ExecutorScheduler.java
@@ -58,7 +58,7 @@ public Disposable scheduleDirect(@NonNull Runnable run) {
         Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
         try {
             if (executor instanceof ExecutorService) {
-                ScheduledDirectTask task = new ScheduledDirectTask(decoratedRun);
+                ScheduledDirectTask task = new ScheduledDirectTask(decoratedRun, interruptibleWorker);
                 Future<?> f = ((ExecutorService)executor).submit(task);
                 task.setFuture(f);
                 return task;
@@ -85,7 +85,7 @@ public Disposable scheduleDirect(@NonNull Runnable run, final long delay, final
         final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
         if (executor instanceof ScheduledExecutorService) {
             try {
-                ScheduledDirectTask task = new ScheduledDirectTask(decoratedRun);
+                ScheduledDirectTask task = new ScheduledDirectTask(decoratedRun, interruptibleWorker);
                 Future<?> f = ((ScheduledExecutorService)executor).schedule(task, delay, unit);
                 task.setFuture(f);
                 return task;
@@ -110,7 +110,7 @@ public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initial
         if (executor instanceof ScheduledExecutorService) {
             Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
             try {
-                ScheduledDirectPeriodicTask task = new ScheduledDirectPeriodicTask(decoratedRun);
+                ScheduledDirectPeriodicTask task = new ScheduledDirectPeriodicTask(decoratedRun, interruptibleWorker);
                 Future<?> f = ((ScheduledExecutorService)executor).scheduleAtFixedRate(task, initialDelay, period, unit);
                 task.setFuture(f);
                 return task;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadWorker.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadWorker.java
@@ -59,7 +59,7 @@ public Disposable schedule(@NonNull final Runnable action, long delayTime, @NonN
      * @return the ScheduledRunnable instance
      */
     public Disposable scheduleDirect(final Runnable run, long delayTime, TimeUnit unit) {
-        ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run));
+        ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run), true);
         try {
             Future<?> f;
             if (delayTime <= 0L) {
@@ -104,7 +104,7 @@ public Disposable schedulePeriodicallyDirect(Runnable run, long initialDelay, lo
 
             return periodicWrapper;
         }
-        ScheduledDirectPeriodicTask task = new ScheduledDirectPeriodicTask(decoratedRun);
+        ScheduledDirectPeriodicTask task = new ScheduledDirectPeriodicTask(decoratedRun, true);
         try {
             Future<?> f = executor.scheduleAtFixedRate(task, initialDelay, period, unit);
             task.setFuture(f);
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTask.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTask.java
@@ -24,8 +24,8 @@ public final class ScheduledDirectPeriodicTask extends AbstractDirectTask implem
 
     private static final long serialVersionUID = 1811839108042568751L;
 
-    public ScheduledDirectPeriodicTask(Runnable runnable) {
-        super(runnable);
+    public ScheduledDirectPeriodicTask(Runnable runnable, boolean interruptOnCancel) {
+        super(runnable, interruptOnCancel);
     }
 
     @Override
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectTask.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectTask.java
@@ -26,8 +26,8 @@ public final class ScheduledDirectTask extends AbstractDirectTask implements Cal
 
     private static final long serialVersionUID = 1811839108042568751L;
 
-    public ScheduledDirectTask(Runnable runnable) {
-        super(runnable);
+    public ScheduledDirectTask(Runnable runnable, boolean interruptOnCancel) {
+        super(runnable, interruptOnCancel);
     }
 
     @Override
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/SingleScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/SingleScheduler.java
@@ -106,7 +106,7 @@ public Worker createWorker() {
     @NonNull
     @Override
     public Disposable scheduleDirect(@NonNull Runnable run, long delay, TimeUnit unit) {
-        ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run));
+        ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run), true);
         try {
             Future<?> f;
             if (delay <= 0L) {
@@ -146,7 +146,7 @@ public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initial
 
             return periodicWrapper;
         }
-        ScheduledDirectPeriodicTask task = new ScheduledDirectPeriodicTask(decoratedRun);
+        ScheduledDirectPeriodicTask task = new ScheduledDirectPeriodicTask(decoratedRun, true);
         try {
             Future<?> f = executor.get().scheduleAtFixedRate(task, initialDelay, period, unit);
             task.setFuture(f);
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/completable/CompletableTimerTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/completable/CompletableTimerTest.java
@@ -37,7 +37,7 @@ public void dispose() {
     public void timerInterruptible() throws Exception {
         ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();
         try {
-            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec) }) {
+            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec, true) }) {
                 final AtomicBoolean interrupted = new AtomicBoolean();
                 TestObserver<Void> to = Completable.timer(1, TimeUnit.MILLISECONDS, s)
                 .doOnComplete(new Action() {
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimerTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimerTest.java
@@ -352,7 +352,7 @@ public void timerDelayZero() {
     public void timerInterruptible() throws Exception {
         ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();
         try {
-            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec) }) {
+            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec, true) }) {
                 final AtomicBoolean interrupted = new AtomicBoolean();
                 TestSubscriber<Long> ts = Flowable.timer(1, TimeUnit.MILLISECONDS, s)
                 .map(new Function<Long, Long>() {
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimerTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimerTest.java
@@ -37,7 +37,7 @@ public void dispose() {
     public void timerInterruptible() throws Exception {
         ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();
         try {
-            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec) }) {
+            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec, true) }) {
                 final AtomicBoolean interrupted = new AtomicBoolean();
                 TestObserver<Long> to = Maybe.timer(1, TimeUnit.MILLISECONDS, s)
                 .map(new Function<Long, Long>() {
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableTimerTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableTimerTest.java
@@ -317,7 +317,7 @@ public void timerDelayZero() {
     public void timerInterruptible() throws Exception {
         ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();
         try {
-            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec) }) {
+            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec, true) }) {
                 final AtomicBoolean interrupted = new AtomicBoolean();
                 TestObserver<Long> to = Observable.timer(1, TimeUnit.MILLISECONDS, s)
                 .map(new Function<Long, Long>() {
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimerTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimerTest.java
@@ -37,7 +37,7 @@ public void disposed() {
     public void timerInterruptible() throws Exception {
         ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();
         try {
-            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec) }) {
+            for (Scheduler s : new Scheduler[] { Schedulers.single(), Schedulers.computation(), Schedulers.newThread(), Schedulers.io(), Schedulers.from(exec, true) }) {
                 final AtomicBoolean interrupted = new AtomicBoolean();
                 TestObserver<Long> to = Single.timer(1, TimeUnit.MILLISECONDS, s)
                 .map(new Function<Long, Long>() {
diff --git a/src/test/java/io/reactivex/rxjava3/internal/schedulers/AbstractDirectTaskTest.java b/src/test/java/io/reactivex/rxjava3/internal/schedulers/AbstractDirectTaskTest.java
@@ -27,7 +27,7 @@ public class AbstractDirectTaskTest extends RxJavaTest {
 
     @Test
     public void cancelSetFuture() {
-        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
             private static final long serialVersionUID = 208585707945686116L;
         };
         final Boolean[] interrupted = { null };
@@ -58,7 +58,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
     @Test
     public void cancelSetFutureCurrentThread() {
-        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
             private static final long serialVersionUID = 208585707945686116L;
         };
         final Boolean[] interrupted = { null };
@@ -91,7 +91,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
     @Test
     public void setFutureCancel() {
-        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
             private static final long serialVersionUID = 208585707945686116L;
         };
         final Boolean[] interrupted = { null };
@@ -119,7 +119,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
     @Test
     public void setFutureCancelSameThread() {
-        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
             private static final long serialVersionUID = 208585707945686116L;
         };
         final Boolean[] interrupted = { null };
@@ -148,7 +148,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
     @Test
     public void finished() {
-        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
             private static final long serialVersionUID = 208585707945686116L;
         };
         final Boolean[] interrupted = { null };
@@ -177,7 +177,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
     @Test
     public void finishedCancel() {
-        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+        AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
             private static final long serialVersionUID = 208585707945686116L;
         };
         final Boolean[] interrupted = { null };
@@ -211,7 +211,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
     @Test
     public void disposeSetFutureRace() {
         for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
-            final AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE) {
+            final AbstractDirectTask task = new AbstractDirectTask(Functions.EMPTY_RUNNABLE, true) {
                 private static final long serialVersionUID = 208585707945686116L;
             };
 
@@ -246,7 +246,7 @@ static class TestDirectTask extends AbstractDirectTask {
         private static final long serialVersionUID = 587679821055711738L;
 
         TestDirectTask() {
-            super(Functions.EMPTY_RUNNABLE);
+            super(Functions.EMPTY_RUNNABLE, true);
         }
     }
 
diff --git a/src/test/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTaskTest.java b/src/test/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTaskTest.java
@@ -35,7 +35,7 @@ public void runnableThrows() {
                 public void run() {
                     throw new TestException();
                 }
-            });
+            }, true);
 
             try {
                 task.run();
diff --git a/src/test/java/io/reactivex/rxjava3/schedulers/ExecutorSchedulerInterruptibleTest.java b/src/test/java/io/reactivex/rxjava3/schedulers/ExecutorSchedulerInterruptibleTest.java

Original file line number	Diff line number	Diff line change
`@@ -32,14 +32,17 @@ abstract class AbstractDirectTask`
`32`	`32`
`33`	`33`	`protected final Runnable runnable;`
`34`	`34`
	`35`	`+ protected final boolean interruptOnCancel;`
	`36`	`+`
`35`	`37`	`protected Thread runner;`
`36`	`38`
`37`	`39`	`protected static final FutureTask<Void> FINISHED = new FutureTask<>(Functions.EMPTY_RUNNABLE, null);`
`38`	`40`
`39`	`41`	`protected static final FutureTask<Void> DISPOSED = new FutureTask<>(Functions.EMPTY_RUNNABLE, null);`
`40`	`42`
`41`		`- AbstractDirectTask(Runnable runnable) {`
	`43`	`+ AbstractDirectTask(Runnable runnable, boolean interruptOnCancel) {`
`42`	`44`	`this.runnable = runnable;`
	`45`	`+ this.interruptOnCancel = interruptOnCancel;`
`43`	`46`	`}`
`44`	`47`
`45`	`48`	`@Override`
`@@ -48,7 +51,7 @@ public final void dispose() {`
`48`	`51`	`if (f != FINISHED && f != DISPOSED) {`
`49`	`52`	`if (compareAndSet(f, DISPOSED)) {`
`50`	`53`	`if (f != null) {`
`51`		`- f.cancel(runner != Thread.currentThread());`
	`54`	`+ cancelFuture(f);`
`52`	`55`	`}`
`53`	`56`	`}`
`54`	`57`	`}`
`@@ -67,7 +70,7 @@ public final void setFuture(Future<?> future) {`
`67`	`70`	`break;`
`68`	`71`	`}`
`69`	`72`	`if (f == DISPOSED) {`
`70`		`- future.cancel(runner != Thread.currentThread());`
	`73`	`+ cancelFuture(future);`
`71`	`74`	`break;`
`72`	`75`	`}`
`73`	`76`	`if (compareAndSet(f, future)) {`
`@@ -76,6 +79,14 @@ public final void setFuture(Future<?> future) {`
`76`	`79`	`}`
`77`	`80`	`}`
`78`	`81`
	`82`	`+ private void cancelFuture(Future<?> future) {`
	`83`	`+ if (runner == Thread.currentThread()) {`
	`84`	`+ future.cancel(false);`
	`85`	`+ } else {`
	`86`	`+ future.cancel(interruptOnCancel);`
	`87`	`+ }`
	`88`	`+ }`
	`89`	`+`
`79`	`90`	`@Override`
`80`	`91`	`public Runnable getWrappedRunnable() {`
`81`	`92`	`return runnable;`
Original file line number	Diff line number	Diff line change
`@@ -24,8 +24,8 @@ public final class ScheduledDirectPeriodicTask extends AbstractDirectTask implem`
`24`	`24`
`25`	`25`	`private static final long serialVersionUID = 1811839108042568751L;`
`26`	`26`
`27`		`- public ScheduledDirectPeriodicTask(Runnable runnable) {`
`28`		`- super(runnable);`
	`27`	`+ public ScheduledDirectPeriodicTask(Runnable runnable, boolean interruptOnCancel) {`
	`28`	`+ super(runnable, interruptOnCancel);`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -26,8 +26,8 @@ public final class ScheduledDirectTask extends AbstractDirectTask implements Cal`
`26`	`26`
`27`	`27`	`private static final long serialVersionUID = 1811839108042568751L;`
`28`	`28`
`29`		`- public ScheduledDirectTask(Runnable runnable) {`
`30`		`- super(runnable);`
	`29`	`+ public ScheduledDirectTask(Runnable runnable, boolean interruptOnCancel) {`
	`30`	`+ super(runnable, interruptOnCancel);`
`31`	`31`	`}`
`32`	`32`
`33`	`33`	`@Override`