Fix TakeLast after removing SingleAssignmentSubscription

- SingleAssignmentSubscription was removed in one merge, TakeLast modified in another merge.

Author: benjchristensen

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. */