diff --git a/src/main/java/rx/internal/schedulers/CachedThreadScheduler.java b/src/main/java/rx/internal/schedulers/CachedThreadScheduler.java index 8e131e58e5..472ade9109 100644 --- a/src/main/java/rx/internal/schedulers/CachedThreadScheduler.java +++ b/src/main/java/rx/internal/schedulers/CachedThreadScheduler.java @@ -204,13 +204,21 @@ public Subscription schedule(Action0 action) { } @Override - public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { + public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) { if (innerSubscription.isUnsubscribed()) { // don't schedule, we are unsubscribed return Subscriptions.unsubscribed(); } - ScheduledAction s = threadWorker.scheduleActual(action, delayTime, unit); + ScheduledAction s = threadWorker.scheduleActual(new Action0() { + @Override + public void call() { + if (isUnsubscribed()) { + return; + } + action.call(); + } + }, delayTime, unit); innerSubscription.add(s); s.addParent(innerSubscription); return s; diff --git a/src/main/java/rx/internal/schedulers/EventLoopsScheduler.java b/src/main/java/rx/internal/schedulers/EventLoopsScheduler.java index afcf7464ed..1eef164937 100644 --- a/src/main/java/rx/internal/schedulers/EventLoopsScheduler.java +++ b/src/main/java/rx/internal/schedulers/EventLoopsScheduler.java @@ -156,20 +156,37 @@ public boolean isUnsubscribed() { } @Override - public Subscription schedule(Action0 action) { + public Subscription schedule(final Action0 action) { if (isUnsubscribed()) { return Subscriptions.unsubscribed(); } - return poolWorker.scheduleActual(action, 0, null, serial); + return poolWorker.scheduleActual(new Action0() { + @Override + public void call() { + if (isUnsubscribed()) { + return; + } + action.call(); + } + }, 0, null, serial); } + @Override - public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { + public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) { if (isUnsubscribed()) { return Subscriptions.unsubscribed(); } - return poolWorker.scheduleActual(action, delayTime, unit, timed); + return poolWorker.scheduleActual(new Action0() { + @Override + public void call() { + if (isUnsubscribed()) { + return; + } + action.call(); + } + }, delayTime, unit, timed); } } @@ -178,4 +195,4 @@ static final class PoolWorker extends NewThreadWorker { super(threadFactory); } } -} \ No newline at end of file +} diff --git a/src/test/java/rx/schedulers/AbstractSchedulerConcurrencyTests.java b/src/test/java/rx/schedulers/AbstractSchedulerConcurrencyTests.java index 2eab100310..9e3fdf94d3 100644 --- a/src/test/java/rx/schedulers/AbstractSchedulerConcurrencyTests.java +++ b/src/test/java/rx/schedulers/AbstractSchedulerConcurrencyTests.java @@ -20,6 +20,8 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -423,4 +425,33 @@ public void call(Integer t) { assertEquals(5, count.get()); } + @Test + public void workerUnderConcurrentUnsubscribeShouldNotAllowLaterTasksToRunDueToUnsubscriptionRace() { + Scheduler scheduler = getScheduler(); + for (int i = 0; i < 1000; i++) { + Worker worker = scheduler.createWorker(); + final Queue q = new ConcurrentLinkedQueue(); + Action0 action1 = new Action0() { + + @Override + public void call() { + q.add(1); + } + }; + Action0 action2 = new Action0() { + + @Override + public void call() { + q.add(2); + } + }; + worker.schedule(action1); + worker.schedule(action2); + worker.unsubscribe(); + if (q.size() == 1 && q.poll() == 2) { + //expect a queue of 1,2 or 1. If queue is just 2 then we have a problem! + fail("wrong order on loop " + i); + } + } + } } diff --git a/src/test/java/rx/schedulers/ExecutorSchedulerTest.java b/src/test/java/rx/schedulers/ExecutorSchedulerTest.java index 0777208cab..ed4e03213d 100644 --- a/src/test/java/rx/schedulers/ExecutorSchedulerTest.java +++ b/src/test/java/rx/schedulers/ExecutorSchedulerTest.java @@ -18,11 +18,9 @@ import static org.junit.Assert.*; import java.lang.management.*; -import java.util.Queue; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; -import org.junit.Assert; import org.junit.Test; import rx.*; @@ -277,32 +275,4 @@ public void call() { assertFalse(w.tasks.hasSubscriptions()); } - - @Test - public void workerUnderConcurrentUnsubscribeShouldNotAllowLaterTasksToRunDueToUnsubscriptionRace() { - Scheduler scheduler = Schedulers.from(Executors.newFixedThreadPool(1)); - for (int i = 0; i< 1000; i++) { - Worker worker = scheduler.createWorker(); - final Queue q = new ConcurrentLinkedQueue(); - Action0 action1 = new Action0() { - - @Override - public void call() { - q.add(1); - }}; - Action0 action2 = new Action0() { - - @Override - public void call() { - q.add(2); - }}; - worker.schedule(action1); - worker.schedule(action2); - worker.unsubscribe(); - if (q.size()==1 && q.poll() == 2) { - //expect a queue of 1,2 or 1. If queue is just 2 then we have a problem! - Assert.fail("wrong order on loop " + i); - } - } - } }