Add take variants to RxScala

Author: zsxwing

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

@@ -799,4 +799,12 @@ class RxScalaDemo extends JUnitSuite {
     assertFalse(o1.sequenceEqual(o3).toBlockingObservable.single)
     assertTrue(o1.sequenceEqual(o4).toBlockingObservable.single)
   }
+
+  @Test def takeExample(): Unit = {
+    val o = (1 to 20).toObservable
+      .zip(Observable.interval(300 millis))
+      .map(_._1)
+      .take(2 seconds)
+    println(o.toBlockingObservable.toList)
+  }
 }

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

@@ -1457,6 +1457,33 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.take(n))
   }
 
+  /**
+   * Returns an Observable that emits those items emitted by source Observable before a specified time runs out.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/take.t.png">
+   *
+   * @param time the length of the time window
+   * @return an Observable that emits those items emitted by the source Observable before the time runs out
+   */
+  def take(time: Duration): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.take(time.length, time.unit))
+  }
+
+  /**
+   * Returns an Observable that emits those items emitted by source Observable before a specified time (on
+   * specified Scheduler) runs out
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/take.ts.png">
+   *
+   * @param time the length of the time window
+   * @param scheduler the Scheduler used for time source
+   * @return an Observable that emits those items emitted by the source Observable before the time runs out,
+   *         according to the specified Scheduler
+   */
+  def take(time: Duration, scheduler: Scheduler) {
+    toScalaObservable[T](asJavaObservable.take(time.length, time.unit, scheduler.asJavaScheduler))
+  }
+
   /**
    * Returns an Observable that emits items emitted by the source Observable so long as a
    * specified condition is true.