Add takeRight(takeLast) variants to RxScala

Author: zsxwing

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -807,4 +807,25 @@ class RxScalaDemo extends JUnitSuite {
       .take(2 seconds)
     println(o.toBlockingObservable.toList)
   }
+
+  @Test def takeRightExample(): Unit = {
+    val o = (1 to 6).toObservable.takeRight(3)
+    assertEquals(List(4, 5, 6), o.toBlockingObservable.toList)
+  }
+
+  @Test def takeRightExample2(): Unit = {
+    val o = (1 to 10).toObservable
+      .zip(Observable.interval(100 millis))
+      .map(_._1)
+      .takeRight(300 millis)
+    println(o.toBlockingObservable.toList)
+  }
+
+  @Test def takeRightExample3(): Unit = {
+    val o = (1 to 10).toObservable
+      .zip(Observable.interval(100 millis))
+      .map(_._1)
+      .takeRight(2, 300 millis)
+    println(o.toBlockingObservable.toList)
+  }
 }

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

@@ -1516,6 +1516,72 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.takeLast(count))
   }
 
+  /**
+   * Return an Observable that emits the items from the source Observable that were emitted in a specified
+   * window of `time` before the Observable completed.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/takeLast.t.png">
+   *
+   * @param time the length of the time window
+   * @return an Observable that emits the items from the source Observable that were emitted in the window of
+   *         time before the Observable completed specified by `time`
+   */
+  def takeRight(time: Duration): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.takeLast(time.length, time.unit))
+  }
+
+  /**
+   * Return an Observable that emits the items from the source Observable that were emitted in a specified
+   * window of `time` before the Observable completed, where the timing information is provided by a specified
+   * Scheduler.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/takeLast.ts.png">
+   *
+   * @param time the length of the time window
+   * @param scheduler the Scheduler that provides the timestamps for the Observed items
+   * @return an Observable that emits the items from the source Observable that were emitted in the window of
+   *         time before the Observable completed specified by `time`, where the timing information is
+   *         provided by `scheduler`
+   */
+  def takeRight(time: Duration, scheduler: Scheduler): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.takeLast(time.length, time.unit, scheduler.asJavaScheduler))
+  }
+
+  /**
+   * Return an Observable that emits at most a specified number of items from the source Observable that were
+   * emitted in a specified window of time before the Observable completed.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/takeLast.tn.png">
+   *
+   * @param count the maximum number of items to emit
+   * @param time the length of the time window
+   * @return an Observable that emits at most `count` items from the source Observable that were emitted
+   *         in a specified window of time before the Observable completed
+   * @throws IllegalArgumentException if `count` is less than zero
+   */
+  def takeRight(count: Int, time: Duration): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.takeLast(count, time.length, time.unit))
+  }
+
+  /**
+   * Return an Observable that emits at most a specified number of items from the source Observable that were
+   * emitted in a specified window of `time` before the Observable completed, where the timing information is
+   * provided by a given Scheduler.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/takeLast.tns.png">
+   *
+   * @param count the maximum number of items to emit
+   * @param time the length of the time window
+   * @param scheduler the Scheduler that provides the timestamps for the observed items
+   * @return an Observable that emits at most `count` items from the source Observable that were emitted
+   *         in a specified window of time before the Observable completed, where the timing information is
+   *         provided by the given `scheduler`
+   * @throws IllegalArgumentException if `count` is less than zero
+   */
+  def takeRight(count: Int, time: Duration, scheduler: Scheduler): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.takeLast(count, time.length, time.unit, scheduler.asJavaScheduler))
+  }
+
   /**
    * Returns an Observable that emits the items from the source Observable only until the
    * `other` Observable emits an item.