Ability to create custom schedulers with behavior based on composing operators.

• rewrote the test to get a little more coverage.
• wrapping each of the onNext/onCompleted to ensure no overlapping calls
• 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

@@ -0,0 +1,263 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.schedulers;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+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.Func1;
+import rx.internal.operators.BufferUntilSubscriber;
+import rx.subjects.PublishSubject;
+import rx.subscriptions.Subscriptions;
+
+/**
+ * Allows the use of operators for controlling the timing around when actions
+ * scheduled on workers are actually done. This makes it possible to layer
+ * additional behavior on an existing {@link Scheduler}. The main parameter is a
+ * function that flattens an {@link Observable} of {@link Completable}s into
+ * just one {@link Completable}. There must be a chain of operators connecting
+ * the returned value to the source {@link Observable} otherwise any work
+ * scheduled on the {@link Scheduler} will not be executed.
+ * <p>
+ * When {@link Scheduler#createWorker()} is invoked a {@link Completable} is
+ * onNext'd to the combinator to be flattened. If the {@link Completable} is not
+ * immediately subscribed to an calls to {@link Worker#schedule} are buffered.
+ * Once the {@link Completable} is subscribed to actions are then scheduled on the
+ * actual {@link Scheduler}. When the {@link Worker} is unsubscribed the
+ * {@link Completable} emits an onComplete and triggers any behavior in the
+ * flattening operator. The {@link Observable} and all {@link Completable}s give
+ * to the flattening function never onError.
+ * <p>
+ * Slowing down the rate to no more than than 1 a second.
+ * 
+ * <pre>
+ * Scheduler slowSched = new ScheduleWhen(workers -> {
+ * 	// use concatenate to make each worker happen one at a time.
+ * 	return Completable.concat(workers.map(worker -> {
+ * 		// delay the starting of the next worker by 1 second.
+ * 		return worker.delay(1, TimeUnit.SECONDS);
+ * 	}));
+ * }, Schedulers.computation());
+ * </pre>
+ * <p>
+ * Limit the amount concurrency two at a time without creating a new fix size
+ * thread pool:
+ * 
+ * <pre>
+ * Scheduler limitSched = new ScheduleWhen(workers -> {
+ * 	// use merge max concurrent to two at a time
+ * 	return Completable.merge(workers, 2);
+ * }, Schedulers.computation());
+ * </pre>
+ */
+@Experimental
+public class ScheduleWhen extends Scheduler {
+	private final Scheduler actualScheduler;
+	private final PublishSubject<Observable<Completable>> workerQueue;
+
+	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();
+		// send it to a custom combinator to pick the order and rate at which
+		// workers are processed.
+		combine.call(workerQueue.onBackpressureBuffer()).subscribe();
+	}
+
+	@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.<ScheduledAction>create();
+		// convert the work of scheduling all the actions into a completable
+		Observable<Completable> actions = actionQueue.map(new Func1<ScheduledAction, Completable>() {
+			@Override
+			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();
+					}
+				});
+			}
+		});
+
+		// a worker that queues the action to the actionQueue subject.
+		Worker worker = new Worker() {
+			private final AtomicBoolean unsubscribed = new AtomicBoolean();
+
+			@Override
+			public void unsubscribe() {
+				// complete the actionQueue when worker is unsubscribed to make
+				// room for the next worker in the workerQueue.
+				if (unsubscribed.compareAndSet(false, true)) {
+					actualWorker.unsubscribe();
+					synchronized (actionQueue) {
+						actionQueue.onCompleted();
+					}
+				}
+			}
+
+			@Override
+			public boolean isUnsubscribed() {
+				return unsubscribed.get();
+			}
+
+			@Override
+			public Subscription schedule(final Action0 action, final long delayTime, final TimeUnit unit) {
+				// send a scheduled action to the actionQueue
+				DelayedAction delayedAction = new DelayedAction(action, delayTime, unit);
+				synchronized (actionQueue) {
+					actionQueue.onNext(delayedAction);
+				}
+				return delayedAction;
+			}
+
+			@Override
+			public Subscription schedule(final Action0 action) {
+				// send a scheduled action to the actionQueue
+				ImmediateAction immediateAction = new ImmediateAction(action);
+				synchronized (actionQueue) {
+					actionQueue.onNext(immediateAction);
+				}
+				return immediateAction;
+			}
+		};
+
+		// enqueue the completable that process actions put in reply subject
+		synchronized (workerQueue) {
+			workerQueue.onNext(actions);
+		}
+
+		// return the worker that adds actions to the reply subject
+		return worker;
+	}
+
+	private static final Subscription SUBSCRIBED = new Subscription() {
+		@Override
+		public void unsubscribe() {
+		}
+
+		@Override
+		public boolean isUnsubscribed() {
+			return false;
+		}
+	};
+
+	private static final Subscription UNSUBSCRIBED = Subscriptions.unsubscribed();
+
+	@SuppressWarnings("serial")
+	private static abstract class ScheduledAction extends AtomicReference<Subscription> implements Subscription {
+		public ScheduledAction() {
+			super(SUBSCRIBED);
+		}
+
+		private final void call(Worker actualWorker) {
+			Subscription oldState = get();
+			// either SUBSCRIBED or UNSUBSCRIBED
+			if (oldState == UNSUBSCRIBED) {
+				// no need to schedule return
+				return;
+			}
+			if (oldState != SUBSCRIBED) {
+				// has already been scheduled return
+				// should not be able to get here but handle it anyway by not
+				// rescheduling.
+				return;
+			}
+
+			Subscription newState = callActual(actualWorker);
+
+			if (!compareAndSet(SUBSCRIBED, newState)) {
+				// set would only fail if the new current state is some other
+				// subscription from a concurrent call to this method.
+				// Unsubscribe from the action just scheduled because it lost
+				// the race.
+				newState.unsubscribe();
+			}
+		}
+
+		protected abstract Subscription callActual(Worker actualWorker);
+
+		@Override
+		public boolean isUnsubscribed() {
+			return get().isUnsubscribed();
+		}
+
+		@Override
+		public void unsubscribe() {
+			Subscription oldState;
+			// no matter what the current state is the new state is going to be
+			Subscription newState = UNSUBSCRIBED;
+			do {
+				oldState = get();
+				if (oldState == UNSUBSCRIBED) {
+					// the action has already been unsubscribed
+					return;
+				}
+			} while (!compareAndSet(oldState, newState));
+
+			if (oldState != SUBSCRIBED) {
+				// the action was scheduled. stop it.
+				oldState.unsubscribe();
+			}
+		}
+	}
+
+	@SuppressWarnings("serial")
+	private static class ImmediateAction extends ScheduledAction {
+		private final Action0 action;
+
+		public ImmediateAction(Action0 action) {
+			this.action = action;
+		}
+
+		@Override
+		protected Subscription callActual(Worker actualWorker) {
+			return actualWorker.schedule(action);
+		}
+	}
+
+	@SuppressWarnings("serial")
+	private static class DelayedAction extends ScheduledAction {
+		private final Action0 action;
+		private final long delayTime;
+		private final TimeUnit unit;
+
+		public DelayedAction(Action0 action, long delayTime, TimeUnit unit) {
+			this.action = action;
+			this.delayTime = delayTime;
+			this.unit = unit;
+		}
+
+		@Override
+		protected Subscription callActual(Worker actualWorker) {
+			return actualWorker.schedule(action, delayTime, unit);
+		}
+	}
+}