Add 'publish' variants to RxScala

Author: zsxwing

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

@@ -297,6 +297,28 @@ class RxScalaDemo extends JUnitSuite {
     shared.connect
   }
 
+  @Test def exampleWithPublish2() {
+    val unshared = Observable.from(1 to 4)
+    val shared = unshared.publish(0)
+    shared.subscribe(n => println(s"subscriber 1 gets $n"))
+    shared.subscribe(n => println(s"subscriber 2 gets $n"))
+    shared.connect
+  }
+
+  @Test def exampleWithPublish3() {
+    val o = Observable.interval(100 millis).take(5).publish((o: Observable[Long]) => o.map(_ * 2))
+    o.subscribe(n => println(s"subscriber 1 gets $n"))
+    o.subscribe(n => println(s"subscriber 2 gets $n"))
+    Thread.sleep(1000)
+  }
+
+  @Test def exampleWithPublish4() {
+    val o = Observable.interval(100 millis).take(5).publish((o: Observable[Long]) => o.map(_ * 2), -1L)
+    o.subscribe(n => println(s"subscriber 1 gets $n"))
+    o.subscribe(n => println(s"subscriber 2 gets $n"))
+    Thread.sleep(1000)
+  }
+
   def doLater(waitTime: Duration, action: () => Unit): Unit = {
     Observable.interval(waitTime).take(1).subscribe(_ => action())
   }

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

@@ -1145,10 +1145,64 @@ trait Observable[+T]
    *
    * @return an [[rx.lang.scala.observables.ConnectableObservable]].
    */
-  def publish: ConnectableObservable[T] = {
+  def publish(): ConnectableObservable[T] = {
     new ConnectableObservable[T](asJavaObservable.publish())
   }
 
+
+  /**
+   * Returns an Observable that emits `initialValue` followed by the items emitted by a `ConnectableObservable` that shares a single subscription to the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/publishConnect.i.png">
+   *
+   * @param initialValue the initial value to be emitted by the resulting Observable
+   * @return a `ConnectableObservable` that shares a single subscription to the underlying Observable and starts with `initialValue`
+   */
+  def publish[U >: T](initialValue: U): ConnectableObservable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    new ConnectableObservable[U](thisJava.publish(initialValue))
+  }
+
+  /**
+   * Returns an Observable that emits the results of invoking a specified selector on items emitted by a `ConnectableObservable`
+   * that shares a single subscription to the underlying sequence.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/publishConnect.f.png">
+   *
+   * @param selector a function that can use the multicasted source sequence as many times as needed, without
+   *                 causing multiple subscriptions to the source sequence. Subscribers to the given source will
+   *                 receive all notifications of the source from the time of the subscription forward.
+   * @return an Observable that emits the results of invoking the selector on the items emitted by a `ConnectableObservable`
+   *         that shares a single subscription to the underlying sequence
+   */
+  def publish[U >: T, R](selector: Observable[U] => Observable[R]): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
+      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+    toScalaObservable[R](thisJava.publish(fJava))
+  }
+
+  /**
+   * Returns an Observable that emits `initialValue` followed by the results of invoking a specified
+   * selector on items emitted by a `ConnectableObservable` that shares a single subscription to the
+   * source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/publishConnect.if.png">
+   *
+   * @param selector a function that can use the multicasted source sequence as many times as needed, without
+   *                 causing multiple subscriptions to the source Observable. Subscribers to the source will
+   *                 receive all notifications of the source from the time of the subscription forward
+   * @param initialValue  the initial value of the underlying `BehaviorSubject`
+   * @return an Observable that emits `initialValue` followed by the results of invoking the selector
+   *         on a `ConnectableObservable` that shares a single subscription to the underlying Observable
+   */
+  def publish[U >: T, R](selector: Observable[U] => Observable[R], initialValue: U): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
+      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+    toScalaObservable[R](thisJava.publish(fJava, initialValue))
+  }
+
   // TODO add Scala-like aggregate function
 
   /**

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

@@ -89,6 +89,10 @@ class CompletenessTest extends JUnitSuite {
       "onExceptionResumeNext(Observable[_ <: T])" -> "onExceptionResumeNext(Observable[U])",
       "parallel(Func1[Observable[T], Observable[R]])" -> "parallel(Observable[T] => Observable[R])",
       "parallel(Func1[Observable[T], Observable[R]], Scheduler)" -> "parallel(Observable[T] => Observable[R], Scheduler)",
+      "publish()" -> "publish()",
+      "publish(T)" -> "publish(U)",
+      "publish(Func1[_ >: Observable[T], _ <: Observable[R]])" -> "publish(Observable[U] => Observable[R])",
+      "publish(Func1[_ >: Observable[T], _ <: Observable[R]], T)" -> "publish(Observable[U] => Observable[R], U)",
       "reduce(Func2[T, T, T])" -> "reduce((U, U) => U)",
       "reduce(R, Func2[R, _ >: T, R])" -> "foldLeft(R)((R, T) => R)",
       "retry()" -> "retry()",