Add flatten variant to RxScala

Author: zsxwing

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

@@ -130,7 +130,7 @@ class RxScalaDemo extends JUnitSuite {
     o.flatten.takeUntil(stopper).toBlockingObservable.foreach(println(_))
   }
 
-  @Test def fattenSomeExample() {
+  @Test def flattenSomeExample() {
     // To merge some observables which are all known already:
     List(
       Observable.interval(200 millis),
@@ -139,6 +139,24 @@ class RxScalaDemo extends JUnitSuite {
     ).toObservable.flatten.take(12).toBlockingObservable.foreach(println(_))
   }
 
+  @Test def flattenExample() {
+    List(
+      Observable.interval(200 millis).map(_ => 1).take(5),
+      Observable.interval(200 millis).map(_ => 2).take(5),
+      Observable.interval(200 millis).map(_ => 3).take(5),
+      Observable.interval(200 millis).map(_ => 4).take(5)
+    ).toObservable.flatten.toBlocking.foreach(println(_))
+  }
+
+  @Test def flattenExample2() {
+    List(
+      Observable.interval(200 millis).map(_ => 1).take(5),
+      Observable.interval(200 millis).map(_ => 2).take(5),
+      Observable.interval(200 millis).map(_ => 3).take(5),
+      Observable.interval(200 millis).map(_ => 4).take(5)
+    ).toObservable.flatten(2).toBlocking.foreach(println(_))
+  }
+
   @Test def rangeAndBufferExample() {
     val o = Observable.from(1 to 18)
     o.buffer(5).subscribe((l: Seq[Int]) => println(l.mkString("[", ", ", "]")))

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

@@ -2329,6 +2329,28 @@ trait Observable[+T]
     toScalaObservable[U](o5)
   }
 
+  /**
+   * Flattens an Observable that emits Observables into a single Observable that emits the items emitted by
+   * those Observables, without any transformation, while limiting the maximum number of concurrent
+   * subscriptions to these Observables.
+   *
+   * <img width="640" height="370" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.oo.png">
+   *
+   * You can combine the items emitted by multiple Observables so that they appear as a single Observable, by
+   * using the `flatten` method.
+   *
+   * @param maxConcurrent the maximum number of Observables that may be subscribed to concurrently
+   * @return an Observable that emits items that are the result of flattening the Observables emitted by the `source` Observable
+   * @throws IllegalArgumentException  if `maxConcurrent` is less than or equal to 0
+   */
+  def flatten[U](maxConcurrent: Int)(implicit evidence: Observable[T] <:< Observable[Observable[U]]): Observable[U] = {
+    val o2: Observable[Observable[U]] = this
+    val o3: Observable[rx.Observable[_ <: U]] = o2.map(_.asJavaObservable)
+    val o4: rx.Observable[_ <: rx.Observable[_ <: U]] = o3.asJavaObservable
+    val o5 = rx.Observable.merge[U](o4, maxConcurrent)
+    toScalaObservable[U](o5)
+  }
+
   /**
    * This behaves like `flatten` except that if any of the merged Observables
    * notify of an error via [[rx.lang.scala.Observer.onError onError]], this method will

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

@@ -191,6 +191,13 @@ class CompletenessTest extends JUnitSuite {
       "just(T, Scheduler)" -> "[use `items(T*).subscribeOn(scheduler)`]",
       "merge(Observable[_ <: T], Observable[_ <: T])" -> "merge(Observable[U])",
       "merge(Observable[_ <: Observable[_ <: T]])" -> "flatten(<:<[Observable[T], Observable[Observable[U]]])",
+      "merge(Observable[_ <: Observable[_ <: T]], Int)" -> "flatten(Int)(<:<[Observable[T], Observable[Observable[U]]])",
+      "merge(Array[Observable[_ <: T]])" -> "[use `Observable.from(array).flatten`]",
+      "merge(Array[Observable[_ <: T]], Scheduler)" -> "[use `Observable.from(array, scheduler).flatten`]",
+      "merge(Iterable[_ <: Observable[_ <: T]])" -> "[use `Observable.from(iter).flatten`]",
+      "merge(Iterable[_ <: Observable[_ <: T]], Int)" -> "[use `Observable.from(iter).flatten(n)`]",
+      "merge(Iterable[_ <: Observable[_ <: T]], Int, Scheduler)" -> "[use `Observable.from(iter, scheduler).flatten(n)]",
+      "merge(Iterable[_ <: Observable[_ <: T]], Scheduler)" -> "[use `Observable.from(iter, scheduler).flatten`]",
       "mergeDelayError(Observable[_ <: T], Observable[_ <: T])" -> "mergeDelayError(Observable[U])",
       "mergeDelayError(Observable[_ <: Observable[_ <: T]])" -> "flattenDelayError(<:<[Observable[T], Observable[Observable[U]]])",
       "parallelMerge(Observable[Observable[T]], Int)" -> "parallelMerge(Int)(<:<[Observable[T], Observable[Observable[U]]])",