Added takeLast to Observable

Author: mairbek

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

@@ -1347,6 +1347,22 @@ public static <T> Observable<T> take(final Observable<T> items, final int num) {
         return _create(OperationTake.take(items, num));
     }
 
+    /**
+     * Returns an Observable that emits the last <code>count</code> items emitted by the source
+     * Observable.
+     *
+     * @param items
+     *            the source Observable
+     * @param count
+     *            the number of items from the end of the sequence emitted by the source
+     *            Observable to emit
+     * @return an Observable that only emits the last <code>count</code> items emitted by the source
+     *         Observable
+     */
+    public static <T> Observable<T> takeLast(final Observable<T> items, final int count) {
+        return _create(OperationTakeLast.takeLast(items, count));
+    }
+
     /**
      * Returns an Observable that emits a single item, a list composed of all the items emitted by
      * the source Observable.
@@ -2235,6 +2251,20 @@ public Observable<T> take(final int num) {
         return take(this, num);
     }
 
+    /**
+     * Returns an Observable that emits the last <code>count</code> items emitted by the source
+     * Observable.
+     *
+     * @param count
+     *            the number of items from the end of the sequence emitted by the source
+     *            Observable to emit
+     * @return an Observable that only emits the last <code>count</code> items emitted by the source
+     *         Observable
+     */
+    public Observable<T> takeLast(final int count) {
+        return takeLast(this, count);
+    }
+
     /**
      * Returns an Observable that emits a single item, a list composed of all the items emitted by
      * the source Observable.

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

@@ -23,13 +23,14 @@
 import rx.util.functions.Func1;
 
 import java.util.Iterator;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
+/**
+ * Returns a specified number of contiguous elements from the end of an observable sequence.
+ */
 public final class OperationTakeLast {
 
     public static <T> Func1<Observer<T>, Subscription> takeLast(final Observable<T> items, final int count) {