Merge pull request #1050 from zsxwing/skip-last

Reimplement the 'SkipLast' operator

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

@@ -81,7 +81,6 @@
 import rx.operators.OperationSample;
 import rx.operators.OperationSequenceEqual;
 import rx.operators.OperationSkip;
-import rx.operators.OperationSkipLast;
 import rx.operators.OperationSkipUntil;
 import rx.operators.OperationSum;
 import rx.operators.OperationSwitch;
@@ -118,6 +117,8 @@
 import rx.operators.OperatorSerialize;
 import rx.operators.OperatorSingle;
 import rx.operators.OperatorSkip;
+import rx.operators.OperatorSkipLast;
+import rx.operators.OperatorSkipLastTimed;
 import rx.operators.OperatorSkipWhile;
 import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorSynchronize;
@@ -6042,7 +6043,7 @@ public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211750.aspx">MSDN: Observable.SkipLast</a>
      */
     public final Observable<T> skipLast(int count) {
-        return create(OperationSkipLast.skipLast(this, count));
+        return lift(new OperatorSkipLast<T>(count));
     }
 
     /**
@@ -6082,7 +6083,7 @@ public final Observable<T> skipLast(long time, TimeUnit unit) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211750.aspx">MSDN: Observable.SkipLast</a>
      */
     public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler) {
-        return create(new OperationSkipLast.SkipLastTimed<T>(this, time, unit, scheduler));
+        return lift(new OperatorSkipLastTimed<T>(time, unit, scheduler));
     }
 
     /**

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

@@ -0,0 +1,77 @@
+/**
+ * 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.ArrayDeque;
+import java.util.Deque;
+
+import rx.Observable.Operator;
+import rx.Subscriber;
+
+/**
+ * Bypasses a specified number of elements at the end of an observable sequence.
+ */
+public class OperatorSkipLast<T> implements Operator<T, T> {
+
+    private final int count;
+
+    public OperatorSkipLast(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 final NotificationLite<T> on = NotificationLite.instance();
+
+            /**
+             * Store the last count elements until now.
+             */
+            private final Deque<Object> deque = new ArrayDeque<Object>();
+
+            @Override
+            public void onCompleted() {
+                subscriber.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                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 emit the value
+                    // directly.
+                    subscriber.onNext(value);
+                    return;
+                }
+                if (deque.size() == count) {
+                    subscriber.onNext(on.getValue(deque.removeFirst()));
+                }
+                deque.offerLast(on.next(value));
+            }
+
+        };
+    }
+
+}