Fix TakeLast after removing SingleAssignmentSubscription

benjchristensen · benjchristensen · commit 4377b364dcb7 · 2013-12-23T21:06:25.000-08:00
- SingleAssignmentSubscription was removed in one merge, TakeLast modified in another merge.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTakeLast.java b/rxjava-core/src/main/java/rx/operators/OperationTakeLast.java
@@ -25,7 +25,7 @@
 import rx.Observer;
 import rx.Scheduler;
 import rx.Subscription;
-import rx.subscriptions.SingleAssignmentSubscription;
+import rx.subscriptions.BooleanSubscription;
 import rx.util.Timestamped;
 
 /**
@@ -156,9 +156,9 @@ public TakeLastTimed(Observable<? extends T> source, int count, long time, TimeU
 
         @Override
         public Subscription onSubscribe(Observer<? super T> t1) {
-            SingleAssignmentSubscription sas = new SingleAssignmentSubscription();
-            sas.set(source.subscribe(new TakeLastTimedObserver<T>(t1, sas, count, ageMillis, scheduler)));
-            return sas;
+            SafeObservableSubscription s = new SafeObservableSubscription();
+            source.subscribe(new TakeLastTimedObserver<T>(t1, s, count, ageMillis, scheduler));
+            return s;
         }
     }
     /** Observes source values and keeps the most recent items. */
