Add combineLatest variant to RxScala

Author: zsxwing

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

@@ -304,6 +304,17 @@ class RxScalaDemo extends JUnitSuite {
     waitFor(combinedCounter)
   }
 
+  @Test def combineLatestExample2() {
+    val firstCounter = Observable.interval(250 millis)
+    val secondCounter = Observable.interval(550 millis)
+    val thirdCounter = Observable.interval(850 millis)
+    val sources = Seq(firstCounter, secondCounter, thirdCounter)
+    val combinedCounter = Observable.combineLatest(sources, (items: Seq[Long]) => items.toList).take(10)
+
+    combinedCounter subscribe {x => println(s"Emitted group: $x")}
+    waitFor(combinedCounter)
+  }
+
   @Test def olympicsExampleWithoutPublish() {
     val medals = Olympics.mountainBikeMedals.doOnEach(_ => println("onNext"))
     medals.subscribe(println(_)) // triggers an execution of medals Observable

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

@@ -4185,6 +4185,26 @@ object Observable {
   def amb[T](sources: Observable[T]*): Observable[T] = {
     toScalaObservable[T](rx.Observable.amb[T](sources.map(_.asJavaObservable).asJava))
   }
+
+  /**
+   * Combines a list of source Observables by emitting an item that aggregates the latest values of each of
+   * the source Observables each time an item is received from any of the source Observables, where this
+   * aggregation is defined by a specified function.
+   *
+   * @tparam T the common base type of source values
+   * @tparam R the result type
+   * @param sources the list of source Observables
+   * @param combineFunction the aggregation function used to combine the items emitted by the source Observables
+   * @return an Observable that emits items that are the result of combining the items emitted by the source
+   *         Observables by means of the given aggregation function
+   */
+  def combineLatest[T, R](sources: Seq[Observable[T]], combineFunction: Seq[T] => R): Observable[R] = {
+    val jSources = new java.util.ArrayList[rx.Observable[_ <: T]](sources.map(_.asJavaObservable).asJava)
+    val jCombineFunction = new rx.functions.FuncN[R] {
+      override def call(args: java.lang.Object*): R = combineFunction(args.map(_.asInstanceOf[T]))
+    }
+    toScalaObservable[R](rx.Observable.combineLatest[T, R](jSources, jCombineFunction))
+  }
 }
 
 

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

@@ -184,6 +184,7 @@ class CompletenessTest extends JUnitSuite {
       "create(OnSubscribeFunc[T])" -> "create(Observer[T] => Subscription)",
       "create(OnSubscribe[T])" -> "apply(Subscriber[T] => Unit)",
       "combineLatest(Observable[_ <: T1], Observable[_ <: T2], Func2[_ >: T1, _ >: T2, _ <: R])" -> "combineLatest(Observable[U])",
+      "combineLatest(List[_ <: Observable[_ <: T]], FuncN[_ <: R])" -> "combineLatest(Seq[Observable[T]], Seq[T] => R)",
       "concat(Observable[_ <: Observable[_ <: T]])" -> "concat(<:<[Observable[T], Observable[Observable[U]]])",
       "defer(Func0[_ <: Observable[_ <: T]])" -> "defer(=> Observable[T])",
       "from(Array[T])" -> "[use `items(T*)`]",