Break up the Worker as Completable into Worker as Observable. Now the schedules actions are the indivisible elements that are subscribed to. The user has additional choice at the cost of making the API more complicated.

Author: abersnaze

src/main/java/rx/schedulers/ScheduleWhen.java

@@ -20,12 +20,13 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Completable;
+import rx.Completable.CompletableOnSubscribe;
+import rx.Completable.CompletableSubscriber;
 import rx.Observable;
 import rx.Scheduler;
 import rx.Subscription;
 import rx.annotations.Experimental;
 import rx.functions.Action0;
-import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.internal.operators.BufferUntilSubscriber;
 import rx.subjects.PublishSubject;
@@ -74,9 +75,9 @@
 @Experimental
 public class ScheduleWhen extends Scheduler {
 	private final Scheduler actualScheduler;
-	private final PublishSubject<Completable> workerQueue;
+	private final PublishSubject<Observable<Completable>> workerQueue;
 
-	public ScheduleWhen(Func1<Observable<Completable>, Completable> combine, Scheduler actualScheduler) {
+	public ScheduleWhen(Func1<Observable<Observable<Completable>>, Completable> combine, Scheduler actualScheduler) {
 		this.actualScheduler = actualScheduler;
 		// workers are converted into completables and put in this queue.
 		this.workerQueue = PublishSubject.create();
@@ -87,18 +88,24 @@ public ScheduleWhen(Func1<Observable<Completable>, Completable> combine, Schedul
 
 	@Override
 	public Worker createWorker() {
+		final Worker actualWorker = actualScheduler.createWorker();
 		// a queue for the actions submitted while worker is waiting to get to
 		// the subscribe to off the workerQueue.
-		final BufferUntilSubscriber<ScheduledAction> actionQueue = BufferUntilSubscriber.create();
-		final Worker actualWorker = actualScheduler.createWorker();
-
+		final BufferUntilSubscriber<ScheduledAction> actionQueue = BufferUntilSubscriber.<ScheduledAction>create();
 		// convert the work of scheduling all the actions into a completable
-		Completable completable = actionQueue.doOnNext(new Action1<ScheduledAction>() {
+		Observable<Completable> actions = actionQueue.map(new Func1<ScheduledAction, Completable>() {
 			@Override
-			public void call(ScheduledAction action) {
-				action.call(actualWorker);
+			public Completable call(final ScheduledAction action) {
+				return Completable.create(new CompletableOnSubscribe() {
+					@Override
+					public void call(CompletableSubscriber actionCompletable) {
+						actionCompletable.onSubscribe(action);
+						action.call(actualWorker);
+						actionCompletable.onCompleted();
+					}
+				});
 			}
-		}).toCompletable();
+		});
 
 		// a worker that queues the action to the actionQueue subject.
 		Worker worker = new Worker() {
@@ -137,7 +144,7 @@ public Subscription schedule(final Action0 action) {
 		};
 
 		// enqueue the completable that process actions put in reply subject
-		workerQueue.onNext(completable);
+		workerQueue.onNext(actions);
 
 		// return the worker that adds actions to the reply subject
 		return worker;

src/test/java/rx/schedulers/ScheduleWhenTest.java

@@ -42,9 +42,15 @@ public Observable<Long> call(Integer t) {
 	public void testMaxConcurrent() {
 		TestScheduler tSched = new TestScheduler();
 
-		Scheduler sched = new ScheduleWhen(new Func1<Observable<Completable>, Completable>() {
+		Scheduler sched = new ScheduleWhen(new Func1<Observable<Observable<Completable>>, Completable>() {
 			@Override
-			public Completable call(Observable<Completable> workers) {
+			public Completable call(Observable<Observable<Completable>> workerActions) {
+				Observable<Completable> workers = workerActions.map(new Func1<Observable<Completable>, Completable>() {
+					@Override
+					public Completable call(Observable<Completable> actions) {
+						return Completable.concat(actions);
+					}
+				});
 				return Completable.merge(workers, 2);
 			}
 		}, tSched);
@@ -70,11 +76,16 @@ public Completable call(Observable<Completable> workers) {
 	public void testDelaySubscription() {
 		final TestScheduler tSched = new TestScheduler();
 
-		Scheduler sched = new ScheduleWhen(new Func1<Observable<Completable>, Completable>() {
+		Scheduler sched = new ScheduleWhen(new Func1<Observable<Observable<Completable>>, Completable>() {
 			@Override
-			public Completable call(Observable<Completable> workers) {
-				return Completable.concat(workers.map(new Func1<Completable, Completable>() {
+			public Completable call(Observable<Observable<Completable>> workerActions) {
+				Observable<Completable> workers = workerActions.map(new Func1<Observable<Completable>, Completable>() {
 					@Override
+					public Completable call(Observable<Completable> actions) {
+						return Completable.concat(actions);
+					}
+				});
+				return Completable.concat(workers.map(new Func1<Completable, Completable>() {
 					public Completable call(Completable worker) {
 						return worker.delay(1, TimeUnit.SECONDS, tSched);
 					}