Merge pull request #1065 from zsxwing/skip-last-timed

Optimize OperatorSkipLastTimed

Author: benjchristensen

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -1300,7 +1300,7 @@ trait Observable[+T]
    * <p>
    * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/skipLast.t.png">
    *
-   * Note: this action will cache all items until "onCompleted" arrives. So don't use it on an infinite Observable.
+   * Note: this action will cache the latest items arriving in the specified time window.
    *
    * @param time the length of the time window
    * @return an Observable that drops those items emitted by the source Observable in a time window before the
@@ -1316,7 +1316,7 @@ trait Observable[+T]
    * <p>
    * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/skipLast.ts.png">
    *
-   * Note: this action will cache all items until "onCompleted" arrives. So don't use it on an infinite Observable.
+   * Note: this action will cache the latest items arriving in the specified time window.
    *
    * @param time the length of the time window
    * @param scheduler the scheduler used as the time source

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

@@ -5567,7 +5567,7 @@ public final Observable<T> skipLast(int count) {
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/skipLast.t.png">
      *
-     * Note: this action will cache all items until "onCompleted" arrives. So don't use it on an infinite Observable.
+     * Note: this action will cache the latest items arriving in the specified time window.
      * 
      * @param time
      *            the length of the time window
@@ -5588,7 +5588,7 @@ public final Observable<T> skipLast(long time, TimeUnit unit) {
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/skipLast.ts.png">
      *
-     * Note: this action will cache all items until "onCompleted" arrives. So don't use it on an infinite Observable.
+     * Note: this action will cache the latest items arriving in the specified time window.
      *
      * @param time
      *            the length of the time window

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

@@ -15,9 +15,8 @@
  */
 package rx.operators;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.ArrayDeque;
+import java.util.Deque;
 import java.util.concurrent.TimeUnit;
 
 import rx.Observable.Operator;
@@ -42,39 +41,37 @@ public OperatorSkipLastTimed(long time, TimeUnit unit, Scheduler scheduler) {
     public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
         return new Subscriber<T>(subscriber) {
 
-            private List<Timestamped<T>> buffer = new ArrayList<Timestamped<T>>();
+            private Deque<Timestamped<T>> buffer = new ArrayDeque<Timestamped<T>>();
+
+            private void emitItemsOutOfWindow(long now) {
+                long limit = now - timeInMillis;
+                while (!buffer.isEmpty()) {
+                    Timestamped<T> v = buffer.getFirst();
+                    if (v.getTimestampMillis() < limit) {
+                        buffer.removeFirst();
+                        subscriber.onNext(v.getValue());
+                    } else {
+                        break;
+                    }
+                }
+            }
 
             @Override
             public void onNext(T value) {
-                buffer.add(new Timestamped<T>(scheduler.now(), value));
+                long now = scheduler.now();
+                emitItemsOutOfWindow(now);
+                buffer.offerLast(new Timestamped<T>(now, value));
             }
 
             @Override
             public void onError(Throwable e) {
-                buffer = Collections.emptyList();
                 subscriber.onError(e);
             }
 
             @Override
             public void onCompleted() {
-                long limit = scheduler.now() - timeInMillis;
-                try {
-                    for (Timestamped<T> v : buffer) {
-                        if (v.getTimestampMillis() < limit) {
-                            try {
-                                subscriber.onNext(v.getValue());
-                            } catch (Throwable t) {
-                                subscriber.onError(t);
-                                return;
-                            }
-                        } else {
-                            break;
-                        }
-                    }
-                    subscriber.onCompleted();
-                } finally {
-                    buffer = Collections.emptyList();
-                }
+                emitItemsOutOfWindow(scheduler.now());
+                subscriber.onCompleted();
             }
 
         };