Avoid self-interrupting of a completing ScheduledAction.

Author: akarnokd

src/main/java/rx/internal/schedulers/NewThreadWorker.java

@@ -67,7 +67,7 @@ public ScheduledAction scheduleActual(final Action0 action, long delayTime, Time
         } else {
             f = executor.schedule(run, delayTime, unit);
         }
-        run.add(Subscriptions.from(f));
+        run.add(f);
 
         return run;
     }

src/main/java/rx/internal/schedulers/ScheduledAction.java

@@ -15,13 +15,15 @@
  */
 package rx.internal.schedulers;
 
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import rx.Subscription;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action0;
 import rx.plugins.RxJavaPlugins;
 import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.Subscriptions;
 
 /**
  * A {@code Runnable} that executes an {@code Action0} and can be cancelled. The analog is the
@@ -33,6 +35,8 @@ public final class ScheduledAction implements Runnable, Subscription {
     volatile int once;
     static final AtomicIntegerFieldUpdater<ScheduledAction> ONCE_UPDATER
             = AtomicIntegerFieldUpdater.newUpdater(ScheduledAction.class, "once");
+    /** Set by the run() method to avoid self interrupting at the end of the run method. */
+    volatile Thread runner;
 
     public ScheduledAction(Action0 action) {
         this.action = action;
@@ -42,6 +46,7 @@ public ScheduledAction(Action0 action) {
     @Override
     public void run() {
         try {
+            runner = Thread.currentThread();
             action.call();
         } catch (Throwable e) {
             // nothing to do but print a System error as this is fatal and there is nowhere else to throw this
@@ -72,15 +77,24 @@ public void unsubscribe() {
     }
 
     /**
-     * @warn javadoc missing
+     * Adds a general Subscription to this {@code ScheduledAction} that will be unsubscribed
+     * if the underlying {@code action} completes or the this scheduled action is cancelled.
      *
-     * @param s
-     * @warn param "s" undescribed
+     * @param s the Subscription to add
      */
     public void add(Subscription s) {
         cancel.add(s);
     }
 
+    /**
+     * Adds the given Future to the unsubscription composite in order to support
+     * cancelling the underlying task in the executor framework.
+     * @param f the future to add
+     */
+    public void add(final Future<?> f) {
+        cancel.add(Subscriptions.create(new FutureCompleter(f)));
+    }
+    
     /**
      * Adds a parent {@link CompositeSubscription} to this {@code ScheduledAction} so when the action is
      * cancelled or terminates, it can remove itself from this parent.
@@ -92,10 +106,32 @@ public void addParent(CompositeSubscription parent) {
         cancel.add(new Remover(this, parent));
     }
 
+    /**
+     * Cancels the captured future if the caller of the call method
+     * is not the same as the runner of the outer ScheduledAction to
+     * prevent unnecessary self-interrupting if the unsubscription
+     * happens from the same thread.
+     */
+    private final class FutureCompleter implements Action0 {
+        private final Future<?> f;
+
+        private FutureCompleter(Future<?> f) {
+            this.f = f;
+        }
+
+        @Override
+        public void call() {
+            if (runner != Thread.currentThread()) {
+                f.cancel(true);
+            }
+        }
+    }
+
     /** Remove a child subscription from a composite when unsubscribing. */
     private static final class Remover implements Subscription {
         final Subscription s;
         final CompositeSubscription parent;
+        @SuppressWarnings("unused")
         volatile int once;
         static final AtomicIntegerFieldUpdater<Remover> ONCE_UPDATER
                 = AtomicIntegerFieldUpdater.newUpdater(Remover.class, "once");

src/test/java/rx/schedulers/NewThreadSchedulerTest.java

@@ -16,8 +16,18 @@
 
 package rx.schedulers;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import junit.framework.Assert;
+
 import org.junit.Test;
+
 import rx.Scheduler;
+import rx.functions.Action0;
+import rx.internal.schedulers.ScheduledAction;
+import rx.subscriptions.Subscriptions;
 
 public class NewThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
 
@@ -35,4 +45,39 @@ public final void testUnhandledErrorIsDeliveredToThreadHandler() throws Interrup
     public final void testHandledErrorIsNotDeliveredToThreadHandler() throws InterruptedException {
         SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(getScheduler());
     }
+    @Test(timeout = 3000)
+    public void testNoSelfInterrupt() throws InterruptedException {
+        Scheduler.Worker worker = Schedulers.newThread().createWorker();
+        
+        final CountDownLatch run = new CountDownLatch(1);
+        final CountDownLatch done = new CountDownLatch(1);
+        final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
+        final AtomicBoolean interruptFlag = new AtomicBoolean();
+        
+        ScheduledAction sa = (ScheduledAction)worker.schedule(new Action0() {
+            @Override
+            public void call() {
+                try {
+                    run.await();
+                } catch (InterruptedException ex) {
+                    exception.set(ex);
+                }
+            }
+        });
+        
+        sa.add(Subscriptions.create(new Action0() {
+            @Override
+            public void call() {
+                interruptFlag.set(Thread.currentThread().isInterrupted());
+                done.countDown();
+            }
+        }));
+        
+        run.countDown();
+        
+        done.await();
+        
+        Assert.assertEquals(null, exception.get());
+        Assert.assertFalse("Interrupted?!", interruptFlag.get());
+    }
 }