Fix "window"; change "amb" parameter to varargs; add examples

Author: zsxwing

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

@@ -156,6 +156,14 @@ class RxScalaDemo extends JUnitSuite {
     ).subscribe(output(_))
   }
 
+  @Test def windowExample2() {
+    val windowObservable = Observable.interval(500 millis)
+    val o = Observable.from(1 to 20).zip(Observable.interval(100 millis)).map(_._1)
+    (for ((o, i) <- o.window(windowObservable).zipWithIndex; n <- o)
+      yield s"Observable#$i emits $n"
+    ).toBlocking.foreach(println)
+  }
+
   @Test def testReduce() {
     assertEquals(10, List(1, 2, 3, 4).toObservable.reduce(_ + _).toBlockingObservable.single)
   }
@@ -731,6 +739,23 @@ class RxScalaDemo extends JUnitSuite {
     println(result)
   }
 
+  @Test def ambWithVarargsExample(): Unit = {
+    val o1 = List(100L, 200L, 300L).toObservable.delay(4 seconds)
+    val o2 = List(1000L, 2000L, 3000L).toObservable.delay(2 seconds)
+    val o3 = List(10000L, 20000L, 30000L).toObservable.delay(4 seconds)
+    val result = Observable.amb(o1, o2, o3).toBlocking.toList
+    println(result)
+  }
+
+  @Test def ambWithSeqExample(): Unit = {
+    val o1 = List(100L, 200L, 300L).toObservable.delay(4 seconds)
+    val o2 = List(1000L, 2000L, 3000L).toObservable.delay(2 seconds)
+    val o3 = List(10000L, 20000L, 30000L).toObservable.delay(4 seconds)
+    val o = Seq(o1, o2, o3)
+    val result = Observable.amb(o: _*).toBlocking.toList
+    println(result)
+  }
+
   @Test def delayExample(): Unit = {
     val o = List(100L, 200L, 300L).toObservable.delay(2 seconds)
     val result = o.toBlockingObservable.toList

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

@@ -670,12 +670,8 @@ trait Observable[+T]
     val func = new Func0[rx.Observable[_ <: Any]]() {
       override def call(): rx.Observable[_ <: Any] = boundary.asJavaObservable
     }
-    val o1: rx.Observable[_ <: rx.Observable[_]] = asJavaObservable.window[Any](func)
-    val o2 = Observable.items(o1).map((x: rx.Observable[_]) => {
-      val x2 = x.asInstanceOf[rx.Observable[_ <: T]]
-      toScalaObservable[T](x2)
-    })
-    o2
+    val jo: rx.Observable[rx.Observable[T]] = asJavaObservable.asInstanceOf[rx.Observable[T]].window[Any](func)
+    toScalaObservable(jo).map(toScalaObservable[T](_))
   }
 
   /**
@@ -3948,8 +3944,8 @@ object Observable {
    * @return an Observable that emits the same sequence of items as whichever of the source Observables
    *         first emitted an item
    */
-  def amb[T](sources: Iterable[Observable[T]]): Observable[T] = {
-    toScalaObservable[T](rx.Observable.amb(sources.map(_.asJavaObservable.asInstanceOf[rx.Observable[T]]).asJava))
+  def amb[T](sources: Observable[T]*): Observable[T] = {
+    toScalaObservable[T](rx.Observable.amb[T](sources.map(_.asJavaObservable).asJava))
   }
 }