Small ObserveOn Improvements

benjchristensen · benjchristensen · commit 28dd5fcdb393 · 2013-12-31T10:10:20.000-08:00
- use Long instead of Int so we don’t overflow
- migrate from deprecated method
diff --git a/rxjava-core/src/main/java/rx/operators/OperationObserveOn.java b/rxjava-core/src/main/java/rx/operators/OperationObserveOn.java
@@ -17,6 +17,7 @@
 
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 import rx.Notification;
 import rx.Observable;
@@ -71,7 +72,7 @@ private class Observation {
             final CompositeSubscription compositeSubscription = new CompositeSubscription();
             final MultipleAssignmentSubscription recursiveSubscription = new MultipleAssignmentSubscription();
             final ConcurrentLinkedQueue<Notification<? extends T>> queue = new ConcurrentLinkedQueue<Notification<? extends T>>();
-            final AtomicInteger counter = new AtomicInteger(0);
+            final AtomicLong counter = new AtomicLong(0);
             private volatile Scheduler recursiveScheduler;
 
             public Observation(Observer<? super T> observer) {
@@ -108,7 +109,7 @@ public Subscription call(Scheduler innerScheduler, T state) {
                 }
 
                 void processQueue() {
-                    recursiveSubscription.setSubscription(recursiveScheduler.schedule(new Action1<Action0>() {
+                    recursiveSubscription.set(recursiveScheduler.schedule(new Action1<Action0>() {
                         @Override
                         public void call(Action0 self) {
                             Notification<? extends T> not = queue.poll();
