Reimplement "SkipLast" operator

Author: zsxwing

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

@@ -6043,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 OperationSkipLast<T>(count));
     }
 
     /**

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

@@ -1,12 +1,12 @@
 /**
  * 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.
@@ -21,10 +21,10 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
 
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
+import rx.Observable.Operator;
 import rx.Observer;
 import rx.Scheduler;
 import rx.Subscriber;
@@ -34,106 +34,60 @@
 /**
  * Bypasses a specified number of elements at the end of an observable sequence.
  */
-public class OperationSkipLast {
+public class OperationSkipLast<T> implements Operator<T, T> {
 
-    /**
-     * Bypasses a specified number of elements at the end of an observable
-     * sequence.
-     * <p>
-     * This operator accumulates a queue with a length enough to store the first
-     * count elements. As more elements are received, elements are taken from
-     * the front of the queue and produced on the result sequence. This causes
-     * elements to be delayed.
-     * 
-     * @param source
-     *            the source sequence.
-     * @param count
-     *            number of elements to bypass at the end of the source
-     *            sequence.
-     * @return An observable sequence containing the source sequence elements
-     *         except for the bypassed ones at the end.
-     * 
-     * @throws IndexOutOfBoundsException
-     *             count is less than zero.
-     */
-    public static <T> OnSubscribeFunc<T> skipLast(
-            Observable<? extends T> source, int count) {
-        return new SkipLast<T>(source, count);
-    }
+    private final int count;
 
-    private static class SkipLast<T> implements OnSubscribeFunc<T> {
-        private final int count;
-        private final Observable<? extends T> source;
-
-        private SkipLast(Observable<? extends T> source, int count) {
-            this.count = count;
-            this.source = source;
+    public OperationSkipLast(int count) {
+        if (count < 0) {
+            throw new IndexOutOfBoundsException("count could not be negative");
         }
+        this.count = count;
+    }
 
-        public Subscription onSubscribe(final Observer<? super T> observer) {
-            if (count < 0) {
-                throw new IndexOutOfBoundsException(
-                        "count could not be negative");
-            }
-            final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(source.unsafeSubscribe(new Subscriber<T>() {
-
-                private final ReentrantLock lock = new ReentrantLock();
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
+        return new Subscriber<T>(subscriber) {
+            /**
+             * Store the last count elements until now.
+             */
+            private final Deque<T> deque = new LinkedList<T>();
 
-                /**
-                 * Store the last count elements until now.
-                 */
-                private final Deque<T> deque = new LinkedList<T>();
+            @Override
+            public void onCompleted() {
+                subscriber.onCompleted();
+            }
 
-                @Override
-                public void onCompleted() {
-                    observer.onCompleted();
-                }
+            @Override
+            public void onError(Throwable e) {
+                subscriber.onError(e);
+            }
 
-                @Override
-                public void onError(Throwable e) {
-                    observer.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;
                 }
-
-                @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.
-                        try {
-                            observer.onNext(value);
-                        } catch (Throwable ex) {
-                            observer.onError(ex);
-                            subscription.unsubscribe();
-                        }
-                        return;
-                    }
-                    lock.lock();
-                    try {
-                        deque.offerLast(value);
-                        if (deque.size() > count) {
-                            // Now deque has count + 1 elements, so the first
-                            // element in the deque definitely does not belong
-                            // to the last count elements of the source
-                            // sequence. We can emit it now.
-                            observer.onNext(deque.removeFirst());
-                        }
-                    } catch (Throwable ex) {
-                        observer.onError(ex);
-                        subscription.unsubscribe();
-                    } finally {
-                        lock.unlock();
-                    }
+                deque.offerLast(value);
+                if (deque.size() > count) {
+                    // Now deque has count + 1 elements, so the first
+                    // element in the deque definitely does not belong
+                    // to the last count elements of the source
+                    // sequence. We can emit it now.
+                    subscriber.onNext(deque.removeFirst());
                 }
+            }
 
-            }));
-        }
+        };
     }
 
     /**
      * Skip delivering values in the time window before the values.
-     * 
+     *
      * @param <T>
      *            the result value type
      */

rxjava-core/src/test/java/rx/operators/OperationSkipLastTest.java

@@ -21,8 +21,8 @@
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static rx.operators.OperationSkipLast.skipLast;
 
+import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
@@ -38,8 +38,7 @@ public class OperationSkipLastTest {
 
     @Test
     public void testSkipLastEmpty() {
-        Observable<String> w = Observable.empty();
-        Observable<String> observable = Observable.create(skipLast(w, 2));
+        Observable<String> observable = Observable.<String>empty().skipLast(2);
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -51,8 +50,7 @@ public void testSkipLastEmpty() {
 
     @Test
     public void testSkipLast1() {
-        Observable<String> w = Observable.from("one", "two", "three");
-        Observable<String> observable = Observable.create(skipLast(w, 2));
+        Observable<String> observable = Observable.from(Arrays.asList("one", "two", "three")).skipLast(2);
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -67,8 +65,7 @@ public void testSkipLast1() {
 
     @Test
     public void testSkipLast2() {
-        Observable<String> w = Observable.from("one", "two");
-        Observable<String> observable = Observable.create(skipLast(w, 2));
+        Observable<String> observable = Observable.from(Arrays.asList("one", "two")).skipLast(2);
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -81,7 +78,7 @@ public void testSkipLast2() {
     @Test
     public void testSkipLastWithZeroCount() {
         Observable<String> w = Observable.from("one", "two");
-        Observable<String> observable = Observable.create(skipLast(w, 0));
+        Observable<String> observable = w.skipLast(0);
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -94,8 +91,7 @@ public void testSkipLastWithZeroCount() {
 
     @Test
     public void testSkipLastWithNull() {
-        Observable<String> w = Observable.from("one", null, "two");
-        Observable<String> observable = Observable.create(skipLast(w, 1));
+        Observable<String> observable = Observable.from(Arrays.asList("one", null, "two")).skipLast(1);
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -107,18 +103,9 @@ public void testSkipLastWithNull() {
         verify(observer, times(1)).onCompleted();
     }
 
-    @Test
+    @Test(expected = IndexOutOfBoundsException.class)
     public void testSkipLastWithNegativeCount() {
-        Observable<String> w = Observable.from("one");
-        Observable<String> observable = Observable.create(skipLast(w, -1));
-
-        @SuppressWarnings("unchecked")
-        Observer<String> observer = mock(Observer.class);
-        observable.subscribe(observer);
-        verify(observer, never()).onNext(any(String.class));
-        verify(observer, times(1)).onError(
-                any(IndexOutOfBoundsException.class));
-        verify(observer, never()).onCompleted();
+        Observable.from("one").skipLast(-1);
     }
 
     @Test