Add buffer variants to RxScala

Author: zsxwing

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

@@ -142,6 +142,12 @@ class RxScalaDemo extends JUnitSuite {
     o.buffer(5).subscribe((l: Seq[Int]) => println(l.mkString("[", ", ", "]")))
   }
 
+  @Test def bufferExample() {
+    val o = Observable.from(1 to 18).zip(Observable.interval(100 millis)).map(_._1)
+    val boundary = Observable.interval(500 millis)
+    o.buffer(boundary).toBlockingObservable.foreach((l: Seq[Int]) => println(l.mkString("[", ", ", "]")))
+  }
+
   @Test def windowExample() {
     (for ((o, i) <- Observable.from(1 to 18).window(5).zipWithIndex; n <- o)
       yield s"Observable#$i emits $n"

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

@@ -565,6 +565,43 @@ trait Observable[+T]
     Observable.jObsOfListToScObsOfSeq(oJava.asInstanceOf[rx.Observable[_ <: java.util.List[T]]])
   }
 
+  /**
+   * Returns an Observable that emits non-overlapping buffered items from the source Observable each time the
+   * specified boundary Observable emits an item.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/buffer8.png">
+   * <p>
+   * Completion of either the source or the boundary Observable causes the returned Observable to emit the
+   * latest buffer and complete.
+   *
+   * @param boundary the boundary Observable
+   * @return an Observable that emits buffered items from the source Observable when the boundary Observable
+   *         emits an item
+   */
+  def buffer(boundary: Observable[Any]): Observable[Seq[T]] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    toScalaObservable(thisJava.buffer(boundary.asJavaObservable)).map(_.asScala)
+  }
+
+  /**
+   * Returns an Observable that emits non-overlapping buffered items from the source Observable each time the
+   * specified boundary Observable emits an item.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/buffer8.png">
+   * <p>
+   * Completion of either the source or the boundary Observable causes the returned Observable to emit the
+   * latest buffer and complete.
+   *
+   * @param boundary the boundary Observable
+   * @param initialCapacity the initial capacity of each buffer chunk
+   * @return an Observable that emits buffered items from the source Observable when the boundary Observable
+   *         emits an item
+   */
+  def buffer(boundary: Observable[Any], initialCapacity: Int): Observable[Seq[T]] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    toScalaObservable(thisJava.buffer(boundary.asJavaObservable, initialCapacity)).map(_.asScala)
+  }
+
   /**
    * Creates an Observable which produces windows of collected values. This Observable produces connected
    * non-overlapping windows. The current window is emitted and replaced with a new window when the