Add "buffer(=> Observable[Any])" to replace "buffer(() => Observable[Any])" and "buffer(Observable[Any])""; Remove the cast for "window(=> Observable[Any])"

Author: zsxwing

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

@@ -408,26 +408,6 @@ trait Observable[+T]
     zip(0 until Int.MaxValue)
   }
 
-  /**
-   * Creates an Observable which produces buffers of collected values.
-   *
-   * This Observable produces connected non-overlapping buffers. The current buffer is
-   * emitted and replaced with a new buffer when the Observable produced by the specified function produces an object. The function will then
-   * be used to create a new Observable to listen for the end of the next buffer.
-   *
-   * @param closings
-   *            The function which is used to produce an [[rx.lang.scala.Observable]] for every buffer created.
-   *            When this [[rx.lang.scala.Observable]] produces an object, the associated buffer
-   *            is emitted and replaced with a new one.
-   * @return
-   *         An [[rx.lang.scala.Observable]] which produces connected non-overlapping buffers, which are emitted
-   *         when the current [[rx.lang.scala.Observable]] created with the function argument produces an object.
-   */
-  def buffer(closings: () => Observable[Any]) : Observable[Seq[T]] = {
-    val f: Func0[_ <: rx.Observable[_ <: Any]] = closings().asJavaObservable
-    val jObs: rx.Observable[_ <: java.util.List[_]] = asJavaObservable.buffer[Any](f)
-    Observable.jObsOfListToScObsOfSeq(jObs.asInstanceOf[rx.Observable[_ <: java.util.List[T]]])
-  }
   /**
    * Creates an Observable which produces buffers of collected values.
    *
@@ -626,13 +606,16 @@ trait Observable[+T]
    * 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 boundary the boundary Observable. Note: This is a by-name parameter,
+   *                 so it is only evaluated when someone subscribes to the returned 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)
+  def buffer(boundary: => Observable[Any]): Observable[Seq[T]] = {
+    val f = new Func0[rx.Observable[_ <: Any]]() {
+      override def call(): rx.Observable[_ <: Any] = boundary.asJavaObservable
+    }
+    toScalaObservable(asJavaObservable.buffer[Any](f)).map(_.asScala)
   }
 
   /**
@@ -670,7 +653,7 @@ trait Observable[+T]
     val func = new Func0[rx.Observable[_ <: Any]]() {
       override def call(): rx.Observable[_ <: Any] = boundary.asJavaObservable
     }
-    val jo: rx.Observable[rx.Observable[T]] = asJavaObservable.asInstanceOf[rx.Observable[T]].window[Any](func)
+    val jo: rx.Observable[_ <: rx.Observable[_ <: T]] = asJavaObservable.window[Any](func)
     toScalaObservable(jo).map(toScalaObservable[T](_))
   }
 

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/CompletenessTest.scala

@@ -72,8 +72,8 @@ class CompletenessTest extends JUnitSuite {
       "all(Func1[_ >: T, Boolean])" -> "forall(T => Boolean)",
       "buffer(Long, Long, TimeUnit)" -> "buffer(Duration, Duration)",
       "buffer(Long, Long, TimeUnit, Scheduler)" -> "buffer(Duration, Duration, Scheduler)",
-      "buffer(Func0[_ <: Observable[_ <: TClosing]])" -> "buffer(() => Observable[Any])",
-      "buffer(Observable[B])" -> "buffer(Observable[Any])",
+      "buffer(Func0[_ <: Observable[_ <: TClosing]])" -> "buffer(=> Observable[Any])",
+      "buffer(Observable[B])" -> "buffer(=> Observable[Any])",
       "buffer(Observable[B], Int)" -> "buffer(Observable[Any], Int)",
       "buffer(Observable[_ <: TOpening], Func1[_ >: TOpening, _ <: Observable[_ <: TClosing]])" -> "buffer(Observable[Opening], Opening => Observable[Any])",
       "contains(Any)" -> "contains(U)",