ensure waiting tasks are cancelled on worker unsubscription (#3867)

davidmoten · zsxwing · commit 3439dd8f73af · 2016-04-20T10:15:37.000-07:00
diff --git a/src/main/java/rx/schedulers/ExecutorScheduler.java b/src/main/java/rx/schedulers/ExecutorScheduler.java
@@ -100,14 +100,17 @@ public void run() {
                     queue.clear();
                     return;
                 }
-
                 ScheduledAction sa = queue.poll();
                 if (sa == null) {
                     return;
                 }
-
                 if (!sa.isUnsubscribed()) {
-                    sa.run();
+                    if (!tasks.isUnsubscribed()) {
+                        sa.run();
+                    } else {
+                        queue.clear();
+                        return;
+                    }
                 }
             } while (wip.decrementAndGet() != 0);
         }

Original file line number	Diff line number	Diff line change
`@@ -100,14 +100,17 @@ public void run() {`
`100`	`100`	`queue.clear();`
`101`	`101`	`return;`
`102`	`102`	`}`
`103`		`-`
`104`	`103`	`ScheduledAction sa = queue.poll();`
`105`	`104`	`if (sa == null) {`
`106`	`105`	`return;`
`107`	`106`	`}`
`108`		`-`
`109`	`107`	`if (!sa.isUnsubscribed()) {`
`110`		`- sa.run();`
	`108`	`+ if (!tasks.isUnsubscribed()) {`
	`109`	`+ sa.run();`
	`110`	`+ } else {`
	`111`	`+ queue.clear();`
	`112`	`+ return;`
	`113`	`+ }`
`111`	`114`	`}`
`112`	`115`	`} while (wip.decrementAndGet() != 0);`
`113`	`116`	`}`