Add 'multicast' variants to RxScala

Author: zsxwing

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

@@ -708,4 +708,19 @@ class RxScalaDemo extends JUnitSuite {
       case e: IllegalArgumentException => println("IllegalArgumentException from skipWithException")
     }
   }
+
+  @Test def multicastExample1(): Unit = {
+    val unshared = Observable.from(1 to 4)
+    val shared = unshared.multicast(Subject())
+    shared.subscribe(n => println(s"subscriber 1 gets $n"))
+    shared.subscribe(n => println(s"subscriber 2 gets $n"))
+    shared.connect
+  }
+
+  @Test def multicastExample2(): Unit = {
+    val unshared = Observable.from(1 to 4)
+    val shared = unshared.multicast[Int, String](() => Subject(), o => o.map("No. " + _))
+    shared.subscribe(n => println(s"subscriber 1 gets $n"))
+    shared.subscribe(n => println(s"subscriber 2 gets $n"))
+  }
 }

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

@@ -220,6 +220,21 @@ trait Observable[+T]
     new ConnectableObservable[R](asJavaObservable.multicast(s))
   }
 
+  /**
+   * Returns an Observable that emits items produced by multicasting the source Observable within a selector function.
+   *
+   * @param subjectFactory the `Subject` factory
+   * @param selector the selector function, which can use the multicasted source Observable subject to the policies
+   *                 enforced by the created `Subject`
+   * @return an Observable that emits the items produced by multicasting the source Observable within a selector function
+   */
+  def multicast[R >: T, U](subjectFactory: () => rx.lang.scala.Subject[R], selector: Observable[R] => Observable[U]): Observable[U] = {
+    val subjectFactoryJava: Func0[rx.subjects.Subject[_ >: T, _ <: R]] = () => subjectFactory().asJavaSubject
+    val selectorJava: Func1[rx.Observable[R], rx.Observable[U]] =
+      (jo: rx.Observable[R]) => selector(toScalaObservable[R](jo)).asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable[U](asJavaObservable.multicast[R, U](subjectFactoryJava, selectorJava))
+  }
+
   /**
    * Returns an Observable that first emits the items emitted by `this`, and then the items emitted
    * by `that`.

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

@@ -140,6 +140,8 @@ class CompletenessTest extends JUnitSuite {
       "merge(Observable[_ <: Observable[_ <: T]])" -> "flatten(<:<[Observable[T], Observable[Observable[U]]])",
       "mergeDelayError(Observable[_ <: T], Observable[_ <: T])" -> "mergeDelayError(Observable[U])",
       "mergeDelayError(Observable[_ <: Observable[_ <: T]])" -> "flattenDelayError(<:<[Observable[T], Observable[Observable[U]]])",
+      "multicast(Subject[_ >: T, _ <: R])" -> "multicast(Subject[R])",
+      "multicast(Func0[_ <: Subject[_ >: T, _ <: TIntermediate]], Func1[_ >: Observable[TIntermediate], _ <: Observable[TResult]])" -> "multicast(() => Subject[R], Observable[R] => Observable[U])",
       "range(Int, Int)" -> "apply(Range)",
       "sequenceEqual(Observable[_ <: T], Observable[_ <: T])" -> "[use `(first zip second) map (p => p._1 == p._2)`]",
       "sequenceEqual(Observable[_ <: T], Observable[_ <: T], Func2[_ >: T, _ >: T, Boolean])" -> "[use `(first zip second) map (p => equality(p._1, p._2))`]",