Add switchMap to RxScala

Author: zsxwing

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

@@ -1228,4 +1228,11 @@ class RxScalaDemo extends JUnitSuite {
     }
     o.subscribe(println(_))
   }
+
+  @Test def switchMapExample() {
+    val o = Observable.interval(300 millis).take(5).switchMap[String] {
+      n => Observable.interval(50 millis).take(10).map(i => s"Seq ${n}: ${i}")
+    }
+    o.toBlocking.foreach(println)
+  }
 }

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

@@ -2237,6 +2237,23 @@ trait Observable[+T]
     )
   }
 
+  /**
+   * Returns a new Observable by applying a function that you supply to each item emitted by the source
+   * Observable that returns an Observable, and then emitting the items emitted by the most recently emitted
+   * of these Observables.
+   *
+   * <img width="640" height="350" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/switchMap.png">
+   *
+   * @param f a function that, when applied to an item emitted by the source Observable, returns an Observable
+   * @return an Observable that emits the items emitted by the Observable returned from applying a function to
+   *         the most recently emitted item emitted by the source Observable
+   */
+  def switchMap[R](f: T => Observable[R]): Observable[R] = {
+    toScalaObservable[R](asJavaObservable.switchMap[R](new Func1[T, rx.Observable[_ <: R]] {
+      def call(t: T): rx.Observable[_ <: R] = f(t).asJavaObservable
+    }))
+  }
+
   /**
    * Given an Observable that emits Observables, creates a single Observable that
    * emits the items emitted by the most recently published of those Observables.