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

Author: abersnaze

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

@@ -0,0 +1,248 @@
+/**
+ * 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.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;
+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<Completable> workerQueue;
+
+	public ScheduleWhen(Func1<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() {
+		// 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();
+
+		// convert the work of scheduling all the actions into a completable
+		Completable completable = actionQueue.doOnNext(new Action1<ScheduledAction>() {
+			@Override
+			public void call(ScheduledAction action) {
+				action.call(actualWorker);
+			}
+		}).toCompletable();
+
+		// 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();
+					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);
+				actionQueue.onNext(delayedAction);
+				return delayedAction;
+			}
+
+			@Override
+			public Subscription schedule(final Action0 action) {
+				// send a scheduled action to the actionQueue
+				ImmediateAction immediateAction = new ImmediateAction(action);
+				actionQueue.onNext(immediateAction);
+				return immediateAction;
+			}
+		};
+
+		// enqueue the completable that process actions put in reply subject
+		workerQueue.onNext(completable);
+
+		// 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);
+		}
+	}
+}

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

@@ -0,0 +1,119 @@
+/**
+ * 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 org.junit.Test;
+
+import rx.Completable;
+import rx.Observable;
+import rx.Scheduler;
+import rx.functions.Func1;
+import rx.observers.TestSubscriber;
+
+public class ScheduleWhenTest {
+	/**
+	 * using the same observable chain but with difference custom schedulers.
+	 */
+	private Observable<Long> work(Scheduler sched) {
+		return Observable.range(1, 5).flatMap(new Func1<Integer, Observable<Long>>() {
+			@Override
+			public Observable<Long> call(Integer t) {
+				return Observable.timer(1, TimeUnit.SECONDS, sched);
+			}
+		});
+	}
+	
+	@Test
+	public void testMaxConcurrent() {
+		TestScheduler tSched = new TestScheduler();
+
+		Scheduler sched = new ScheduleWhen(new Func1<Observable<Completable>, Completable>() {
+			@Override
+			public Completable call(Observable<Completable> workers) {
+				return Completable.merge(workers, 2);
+			}
+		}, tSched);
+
+		TestSubscriber<Long> tSub = TestSubscriber.create();
+
+		work(sched).subscribe(tSub);
+
+		tSub.assertValueCount(0);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(2);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(4);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(5);
+		tSub.assertCompleted();
+	}
+
+	@Test
+	public void testDelaySubscription() {
+		TestScheduler tSched = new TestScheduler();
+
+		Scheduler sched = new ScheduleWhen(new Func1<Observable<Completable>, Completable>() {
+			@Override
+			public Completable call(Observable<Completable> workers) {
+				return Completable.concat(workers.map(new Func1<Completable, Completable>() {
+					@Override
+					public Completable call(Completable worker) {
+						return worker.delay(1, TimeUnit.SECONDS, tSched);
+					}
+				}));
+			}
+		}, tSched);
+
+		TestSubscriber<Long> tSub = TestSubscriber.create();
+
+		work(sched).subscribe(tSub);
+
+		tSub.assertValueCount(0);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(1);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(1);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(2);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(2);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(3);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(3);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(4);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(4);
+
+		tSched.advanceTimeBy(1, TimeUnit.SECONDS);
+		tSub.assertValueCount(5);
+		tSub.assertCompleted();
+	}
+}