OperatorTakeLast

Author: zsxwing

rxjava-core/src/main/java/rx/Observable.java

@@ -6356,7 +6356,7 @@ public final Observable<T> takeFirst(Func1<? super T, Boolean> predicate) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-takelast">RxJava Wiki: takeLast()</a>
      */
     public final Observable<T> takeLast(final int count) {
-        return create(OperationTakeLast.takeLast(this, count));
+        return lift(new OperatorTakeLast<T>(count));
     }
 
     /**
@@ -6400,10 +6400,7 @@ public final Observable<T> takeLast(int count, long time, TimeUnit unit) {
      *             if {@code count} is less than zero
      */
     public final Observable<T> takeLast(int count, long time, TimeUnit unit, Scheduler scheduler) {
-        if (count < 0) {
-            throw new IllegalArgumentException("count >= 0 required");
-        }
-        return create(OperationTakeLast.takeLast(this, count, time, unit, scheduler));
+        return lift(new OperatorTakeLastTimed<T>(count, time, unit, scheduler));
     }
 
     /**
@@ -6441,7 +6438,7 @@ public final Observable<T> takeLast(long time, TimeUnit unit) {
      *         provided by {@code scheduler}
      */
     public final Observable<T> takeLast(long time, TimeUnit unit, Scheduler scheduler) {
-        return create(OperationTakeLast.takeLast(this, time, unit, scheduler));
+        return lift(new OperatorTakeLastTimed<T>(time, unit, scheduler));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationTakeLast.java

@@ -0,0 +1,85 @@
+/**
+ * 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.Observable.Operator;
+import rx.Subscriber;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/**
+ * Returns an Observable that emits the last <code>count</code> items emitted by the source
+ * Observable.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/last.png">
+ */
+public final class OperatorTakeLast<T> implements Operator<T, T> {
+
+    private final int count;
+
+    public OperatorTakeLast(int count) {
+        if (count < 0) {
+            throw new IndexOutOfBoundsException("count could not be negative");
+        }
+        this.count = count;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
+        return new Subscriber<T>(subscriber) {
+
+            private NotificationLite<T> notification = NotificationLite.instance();
+            /**
+             * Store the last count elements until now.
+             */
+            private Deque<Object> deque = new ArrayDeque<Object>();
+
+            @Override
+            public void onCompleted() {
+                try {
+                    for (Object value : deque) {
+                        subscriber.onNext(notification.getValue(value));
+                    }
+                } catch (Throwable e) {
+                    onError(e);
+                    return;
+                }
+                deque.clear();
+                subscriber.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                deque.clear();
+                subscriber.onError(e);
+            }
+
+            @Override
+            public void onNext(T value) {
+                if (count == 0) {
+                    // If count == 0, we do not need to put value into deque and
+                    // remove it at once. We can ignore the value directly.
+                    return;
+                }
+                if (deque.size() == count) {
+                    deque.removeFirst();
+                }
+                deque.offerLast(notification.next(value));
+            }
+        };
+    }
+}