add multicast, publish, replay

Author: samuelgruetter

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

@@ -20,7 +20,6 @@ package rx.lang.scala
 import org.scalatest.junit.JUnitSuite
 import scala.collection.Seq
 import rx.lang.scala.observables.BlockingObservable
-import rx.lang.scala.observables.ConnectableObservable
 
 
 /**
@@ -38,6 +37,7 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
   import rx.util.functions._
   import rx.lang.scala.{Notification, Subscription, Scheduler, Observer}
   import rx.lang.scala.util._
+  import rx.lang.scala.subjects.Subject
   import rx.lang.scala.ImplicitFunctionConversions._
 
   /**
@@ -132,11 +132,13 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
    *            into
    * @param <R>
    *            result type
-   * @return a {@link ConnectableObservable} that upon connection causes the source Observable to
-   *         push results into the specified {@link Subject}
+   * @return a pair of a start function and an {@link Observable} such that when the start function
+   *         is called, the Observable starts to push results into the specified {@link Subject}
    */
-  // public <R> ConnectableObservable<R> multicast(Subject<T, R> subject) TODO
-  
+  def multicast[R](subject: Subject[T, R]): (() => Subscription, Observable[R]) = {
+    val javaCO = asJava.multicast[R](subject)
+    (() => javaCO.connect(), Observable[R](javaCO))
+  }
 
   /**
    * Returns an Observable that first emits the items emitted by this, and then the items emitted
@@ -904,11 +906,12 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
    * <p>
    * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/replay.png">
    * 
-   * @return a {@link ConnectableObservable} that upon connection causes the source Observable to
-   *         emit items to its {@link Observer}s
+   * @return a pair of a start function and an {@link Observable} such that when the start function
+   *         is called, the Observable starts to emit items to its {@link Observer}s
    */
-  def replay(): ConnectableObservable[T] = {
-    new ConnectableObservable[T](asJava.replay())
+  def replay(): (() => Subscription, Observable[T]) = {
+    val javaCO = asJava.replay()
+    (() => javaCO.connect(), Observable[T](javaCO))
   }
 
   /**
@@ -937,11 +940,12 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
    * <p>
    * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/publishConnect.png">
    * 
-   * @return a {@link ConnectableObservable} that upon connection causes the source Observable to
-   *         emit items to its {@link Observer}s
+   * @return a pair of a start function and an {@link Observable} such that when the start function
+   *         is called, the Observable starts to emit items to its {@link Observer}s
    */
-  def publish: ConnectableObservable[T] = {
-    new ConnectableObservable[T](asJava.publish())
+  def publish: (() => Subscription, Observable[T]) = {
+    val javaCO = asJava.publish()
+    (() => javaCO.connect(), Observable[T](javaCO))
   }
 
   // There is no aggregate function with signature

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

@@ -191,6 +191,44 @@ class RxScalaDemo extends JUnitSuite {
     waitFor(firstMedalOfEachCountry)
   }
 
+  @Test def exampleWithoutPublish() {
+    val unshared = Observable(1 to 4)
+    unshared.subscribe(n => println(s"subscriber 1 gets $n"))
+    unshared.subscribe(n => println(s"subscriber 2 gets $n"))
+  }
+  
+  @Test def exampleWithPublish() {
+    val unshared = Observable(1 to 4)
+    val (startFunc, shared) = unshared.publish
+    shared.subscribe(n => println(s"subscriber 1 gets $n"))
+    shared.subscribe(n => println(s"subscriber 2 gets $n"))
+    startFunc()
+  }
+  
+  def doLater(waitTime: Duration, action: () => Unit): Unit = {
+    Observable.interval(waitTime).take(1).subscribe(_ => action())
+  }
+  
+  @Test def exampleWithoutReplay() {
+    val numbers = Observable.interval(1000 millis).take(6)
+    val (startFunc, sharedNumbers) = numbers.publish
+    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
+    startFunc()
+    // subscriber 2 misses 0, 1, 2!
+    doLater(3500 millis, () => { sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n")) })
+    waitFor(sharedNumbers)
+  }
+  
+  @Test def exampleWithReplay() {
+    val numbers = Observable.interval(1000 millis).take(6)
+    val (startFunc, sharedNumbers) = numbers.replay
+    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
+    startFunc()
+    // subscriber 2 subscribes later but still gets all numbers
+    doLater(3500 millis, () => { sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n")) })
+    waitFor(sharedNumbers)
+  }
+  
   def output(s: String): Unit = println(s)
 
   // blocks until obs has completed

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/observables/ConnectableObservable.scala

@@ -53,12 +53,6 @@ rx.plugins.RxJavaErrorHandler
 rx.plugins.RxJavaObservableExecutionHook
 rx.plugins.RxJavaPlugins
 
-rx.subjects.AsyncSubject
-rx.subjects.BehaviorSubject
-rx.subjects.PublishSubject
-rx.subjects.ReplaySubject
-rx.subjects.Subject
-
 rx.subscriptions.BooleanSubscription
 rx.subscriptions.CompositeSubscription
 rx.subscriptions.Subscriptions

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

@@ -0,0 +1,14 @@
+package rx.lang.scala
+
+package object subjects {
+
+  // in Java: public abstract class Subject<T,R> extends Observable<R> implements Observer<T>
+  type Subject[-T, +R] = rx.subjects.Subject[_ >: T, _ <: R]
+
+  // TODO (including static methods of these classes)
+  // rx.subjects.AsyncSubject
+  // rx.subjects.BehaviorSubject
+  // rx.subjects.PublishSubject
+  // rx.subjects.ReplaySubject
+
+}