ObserveOn Merge from @akarnokd:OperatorRepeat2

akarnokd · benjchristensen · commit 9e2691729d94 · 2014-02-08T14:23:21.000-08:00
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
@@ -49,7 +49,6 @@
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
 import rx.operators.OperationDistinctUntilChanged;
-import rx.operators.OperatorDoOnEach;
 import rx.operators.OperationElementAt;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
@@ -63,13 +62,11 @@
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMinMax;
 import rx.operators.OperationMulticast;
-import rx.operators.OperationObserveOn;
 import rx.operators.OperationOnErrorResumeNextViaFunction;
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallelMerge;
-import rx.operators.OperatorRepeat;
 import rx.operators.OperationReplay;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
@@ -96,18 +93,21 @@
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationUsing;
 import rx.operators.OperationWindow;
-import rx.operators.OperatorSubscribeOn;
-import rx.operators.OperatorZip;
 import rx.operators.OperatorCast;
+import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorFromIterable;
 import rx.operators.OperatorGroupBy;
 import rx.operators.OperatorMap;
 import rx.operators.OperatorMerge;
+import rx.operators.OperatorObserveOn;
 import rx.operators.OperatorParallel;
+import rx.operators.OperatorRepeat;
+import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorTake;
 import rx.operators.OperatorTimestamp;
 import rx.operators.OperatorToObservableList;
 import rx.operators.OperatorToObservableSortedList;
+import rx.operators.OperatorZip;
 import rx.operators.OperatorZipIterable;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
@@ -5151,7 +5151,7 @@ public final <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-observeon">RxJava Wiki: observeOn()</a>
      */
     public final Observable<T> observeOn(Scheduler scheduler) {
-        return create(OperationObserveOn.observeOn(this, scheduler));
+        return lift(new OperatorObserveOn<T>(scheduler));
     }
 
     /**
diff --git a/rxjava-core/src/main/java/rx/operators/OperationObserveOn.java b/rxjava-core/src/main/java/rx/operators/OperationObserveOn.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperatorObserveOn.java b/rxjava-core/src/main/java/rx/operators/OperatorObserveOn.java
@@ -0,0 +1,67 @@
+/**
+ * Copyright 2014 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.operators;
+
+import rx.Notification;
+import rx.Scheduler;
+import rx.Subscriber;
+import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Action0;
+
+/**
+ * Move the observation of events to another thread via Scheduler.
+ * @param <T> the item type
+ */
+public class OperatorObserveOn<T> implements Operator<T, T> {
+    final Scheduler scheduler;
+    public OperatorObserveOn(Scheduler scheduler) {
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> t1) {
+        final QueueDrain qd = new QueueDrain(t1);
+        final CompositeSubscription csub = new CompositeSubscription();
+        t1.add(csub);
+        return new Subscriber<T>(t1) {
+            /** Dispatch the notification value. */
+            void run(final Notification<T> nt) {
+                qd.enqueue(new Action0() {
+                    @Override
+                    public void call() {
+                        nt.accept(t1);
+                    }
+                });
+                qd.tryDrainAsync(scheduler, csub);
+            }
+            @Override
+            public void onNext(final T args) {
+                run(Notification.createOnNext(args));
+            }
+
+            @Override
+            public void onError(final Throwable e) {
+                run(Notification.<T>createOnError(e));
+            }
+
+            @Override
+            public void onCompleted() {
+                run(Notification.<T>createOnCompleted());
+            }
+        };
+    }
+    
+}
diff --git a/rxjava-core/src/main/java/rx/operators/QueueDrain.java b/rxjava-core/src/main/java/rx/operators/QueueDrain.java
@@ -0,0 +1,106 @@
+/**
+ * Copyright 2014 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.operators;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.MultipleAssignmentSubscription;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+
+/**
+ * Action queue ensuring that only a single drain caller succeeds at a time.
+ * This class can be used to execute work without the issues of reentrancy and
+ * concurrency.
+ */
+public final class QueueDrain implements Runnable, Action0 {
+    /** The number of work items. */
+    private final AtomicInteger wip = new AtomicInteger();
+    /** The action queue. */
+    private final BlockingQueue<Action0> queue = new LinkedBlockingQueue<Action0>();
+    /** The subscription to stop the queue processing. */
+    private final Subscription k;
+    /**
+     * Constructor which takes a cancellation token.
+     * @param k the cancellation token (aka subscription).
+     */
+    public QueueDrain(Subscription k) {
+        this.k = k;
+    }
+    /**
+     * Enqueue an action.
+     * To execute any queued action, call {@link #tryDrain()} or
+     * submit this instance to a {@code Scheduler.schedule()} method.
+     * @param action the action to enqueue, not null
+     */
+    public void enqueue(Action0 action) {
+        if (!k.isUnsubscribed()) {
+            queue.add(action);
+        }
+    }
+    /**
+     * Try draining the queue and executing the actions in it.
+     */
+    public void tryDrain() {
+        if (wip.incrementAndGet() > 1 || k.isUnsubscribed()) {
+            return;
+        }
+        do {
+            queue.poll().call();
+        } while (wip.decrementAndGet() > 0 && !k.isUnsubscribed());
+    }
+    /**
+     * Try draining the queue on the given scheduler.
+     * The method ensures that only one thread is actively draining the
+     * queue on the given scheduler.
+     * @param scheduler the scheduler where the draining should happen
+     * @param cs the composite subscription to track the schedule
+     */
+    public void tryDrainAsync(Scheduler scheduler, final CompositeSubscription cs) {
+        if (cs.isUnsubscribed() || wip.incrementAndGet() > 1) {
+            return;
+        }
+        // add tracking subscription only if schedule is run to avoid overfilling cs
+        final MultipleAssignmentSubscription mas = new MultipleAssignmentSubscription();
+        cs.add(mas);
+        mas.set(scheduler.schedule(new Action1<Scheduler.Inner>() {
+            @Override
+            public void call(Scheduler.Inner o) {
+                if (!cs.isUnsubscribed()) {
+                    do {
+                        queue.poll().call();
+                    } while (wip.decrementAndGet() > 0 && !cs.isUnsubscribed());
+                }
+                cs.remove(mas);
+            }
+        }));
+    }
+    @Override
+    public void run() {
+        // to help the draining of the queue on a ThreadPool/Scheduler
+        tryDrain();
+    }
+
+    @Override
+    public void call() {
+        tryDrain();
+    }
+    
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperatorObserveOnTest.java b/rxjava-core/src/test/java/rx/operators/OperatorObserveOnTest.java
@@ -18,7 +18,6 @@
 import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
-import static rx.operators.OperationObserveOn.*;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -37,7 +36,7 @@
 import rx.util.functions.Action1;
 import rx.util.functions.Func1;
 
-public class OperationObserveOnTest {
+public class OperatorObserveOnTest {
 
     /**
      * This is testing a no-op path since it uses Schedulers.immediate() which will not do scheduling.
@@ -46,7 +45,7 @@ public class OperationObserveOnTest {
     @SuppressWarnings("unchecked")
     public void testObserveOn() {
         Observer<Integer> observer = mock(Observer.class);
-        Observable.create(observeOn(Observable.from(1, 2, 3), Schedulers.immediate())).subscribe(observer);
+        Observable.from(1, 2, 3).observeOn(Schedulers.immediate()).subscribe(observer);
 
         verify(observer, times(1)).onNext(1);
         verify(observer, times(1)).onNext(2);
diff --git a/rxjava-core/src/test/java/rx/operators/OperatorTakeTest.java b/rxjava-core/src/test/java/rx/operators/OperatorTakeTest.java
