TrampolineScheduler & Unsubscribe

benjchristensen · benjchristensen · commit 7058a7674801 · 2014-06-04T16:09:16.000-07:00
Unsubscribing should prevent new additions to a Worker, but not prevent already scheduled work, and definitely not affect other Workers using the same thread (by modifying the ThreadLocal as it was doing).

See the unit test for details of how unsubscribing 1 Worker could prevent work from being done on a completely separate Worker.
diff --git a/rxjava-core/src/main/java/rx/schedulers/TrampolineScheduler.java b/rxjava-core/src/main/java/rx/schedulers/TrampolineScheduler.java
@@ -83,9 +83,6 @@ private Subscription enqueue(Action0 action, long execTime) {
 
             if (exec) {
                 while (!queue.isEmpty()) {
-                    if (innerSubscription.isUnsubscribed()) {
-                        return Subscriptions.empty();
-                    }
                     queue.poll().action.call();
                 }
 
@@ -108,7 +105,6 @@ public void call() {
 
         @Override
         public void unsubscribe() {
-            QUEUE.set(null); // this assumes we are calling unsubscribe from the same thread
             innerSubscription.unsubscribe();
         }
 
diff --git a/rxjava-core/src/test/java/rx/schedulers/TrampolineSchedulerTest.java b/rxjava-core/src/test/java/rx/schedulers/TrampolineSchedulerTest.java
@@ -15,12 +15,18 @@
  */
 package rx.schedulers;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Scheduler;
+import rx.Scheduler.Worker;
+import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
 
@@ -55,4 +61,62 @@ public void call(String t) {
             }
         });
     }
+
+    @Test
+    public void testNestedTrampolineWithUnsubscribe() {
+        final ArrayList<String> workDone = new ArrayList<String>();
+        Worker worker = Schedulers.trampoline().createWorker();
+        worker.schedule(new Action0() {
+
+            @Override
+            public void call() {
+                doWorkOnNewTrampoline("A", workDone);
+            }
+
+        });
+
+        final Worker worker2 = Schedulers.trampoline().createWorker();
+        worker2.schedule(new Action0() {
+
+            @Override
+            public void call() {
+                doWorkOnNewTrampoline("B", workDone);
+                // we unsubscribe worker2 ... it should not affect work scheduled on a separate Trampline.Worker
+                worker2.unsubscribe();
+            }
+
+        });
+
+        assertEquals(6, workDone.size());
+        assertEquals(Arrays.asList("A.1", "A.B.1", "A.B.2", "B.1", "B.B.1", "B.B.2"), workDone);
+    }
+
+    private static void doWorkOnNewTrampoline(final String key, final ArrayList<String> workDone) {
+        Worker worker = Schedulers.trampoline().createWorker();
+        worker.schedule(new Action0() {
+
+            @Override
+            public void call() {
+                String msg = key + ".1";
+                workDone.add(msg);
+                System.out.println(msg);
+                Worker worker3 = Schedulers.trampoline().createWorker();
+                worker3.schedule(createPrintAction(key + ".B.1", workDone));
+                worker3.schedule(createPrintAction(key + ".B.2", workDone));
+            }
+
+        });
+    }
+
+    private static Action0 createPrintAction(final String message, final ArrayList<String> workDone) {
+        return new Action0() {
+
+            @Override
+            public void call() {
+                System.out.println(message);
+                workDone.add(message);
+            }
+
+        };
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -83,9 +83,6 @@ private Subscription enqueue(Action0 action, long execTime) {`
`83`	`83`
`84`	`84`	`if (exec) {`
`85`	`85`	`while (!queue.isEmpty()) {`
`86`		`- if (innerSubscription.isUnsubscribed()) {`
`87`		`- return Subscriptions.empty();`
`88`		`- }`
`89`	`86`	`queue.poll().action.call();`
`90`	`87`	`}`
`91`	`88`
`@@ -108,7 +105,6 @@ public void call() {`
`108`	`105`
`109`	`106`	`@Override`
`110`	`107`	`public void unsubscribe() {`
`111`		`- QUEUE.set(null); // this assumes we are calling unsubscribe from the same thread`
`112`	`108`	`innerSubscription.unsubscribe();`
`113`	`109`	`}`
`114`	`110`