add Subscriber and Observable creation from Subscriber[T] => Unit

Author: samuelgruetter

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

@@ -463,15 +463,43 @@ class RxScalaDemo extends JUnitSuite {
     List("a", "b").toObservable.zipWithIndex.takeWhile{case (elem, index) => condition}.map(_._1)
   }
 
-  @Test def createExample() {
+  def calculateElement(index: Int): String = {
+    println("omg I'm calculating so hard")
+    index match {
+      case 0 => "a"
+      case 1 => "b"
+      case _ => throw new IllegalArgumentException
+    }
+  }
+  
+  /**
+   * This is a bad way of using Observable.create, because even if the consumer unsubscribes,
+   * all elements are calculated.
+   */
+  @Test def createExampleBad() {
     val o = Observable.create[String](observer => {
-      // this is bad because you cannot unsubscribe!
-      observer.onNext("a")
-      observer.onNext("b")
+      observer.onNext(calculateElement(0))
+      observer.onNext(calculateElement(1))
       observer.onCompleted()
       Subscription {}
     })
-    o.subscribe(println(_))
+    o.take(1).subscribe(println(_))
+  }
+  
+  /**
+   * This is the good way of doing it: If the consumer unsubscribes, no more elements are 
+   * calculated.
+   */
+  @Test def createExampleGood() {
+    val o = Observable[String](subscriber => {
+      var i = 0
+      while (i < 2 && !subscriber.isUnsubscribed) {
+        subscriber.onNext(calculateElement(i))
+        i += 1
+      }
+      if (!subscriber.isUnsubscribed) subscriber.onCompleted()
+    })
+    o.take(1).subscribe(println(_))
   }
 
   def output(s: String): Unit = println(s)

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

@@ -17,10 +17,8 @@ package rx.lang.scala
 
 import java.lang.Exception
 import java.{ lang => jlang }
-
 import scala.language.implicitConversions
 import scala.collection.Seq
-
 import rx.functions._
 import rx.lang.scala.JavaConversions._
 
@@ -56,6 +54,13 @@ object ImplicitFunctionConversions {
       }
     }
 
+  implicit def scalaAction1ToOnSubscribe[T](f: Subscriber[T] => Unit) =
+    new rx.Observable.OnSubscribe[T] {
+      def call(s: rx.Subscriber[_ >: T]): Unit = {
+        f(s)
+      }
+    }
+  
   implicit def scalaByNameParamToFunc0[B](param: => B): Func0[B] =
     new Func0[B] {
       def call(): B = param

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

@@ -32,6 +32,10 @@ object JavaConversions {
 
   implicit def toScalaSubscription(s: rx.Subscription): Subscription = Subscription(s)
 
+  implicit def toJavaSubscriber[T](s: Subscriber[T]): rx.Subscriber[_ >: T] = s.asJavaSubscriber
+  
+  implicit def toScalaSubscriber[T](s: rx.Subscriber[_ >: T]): Subscriber[T] = Subscriber(s)
+  
   implicit def scalaSchedulerToJavaScheduler(s: Scheduler): rx.Scheduler = s.asJavaScheduler
   implicit def javaSchedulerToScalaScheduler(s: rx.Scheduler): Scheduler = Scheduler(s)
 

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

@@ -2272,9 +2272,6 @@ object Observable {
    * should invoke the Observer's [[rx.lang.scala.Observer.onNext onNext]], [[rx.lang.scala.Observer.onError onError]], and [[rx.lang.scala.Observer.onCompleted onCompleted]] methods
    * appropriately.
    *
-   * A well-formed Observable must invoke either the Observer's `onCompleted` method
-   * exactly once or its `onError` method exactly once.
-   *
    * See <a href="http://go.microsoft.com/fwlink/?LinkID=205219">Rx Design Guidelines (PDF)</a>
    * for detailed information.
    *
@@ -2296,6 +2293,41 @@ object Observable {
     }))
   }
 
+  /*
+  Note: It's dangerous to have two overloads where one takes an `Observer[T] => Subscription`
+  function and the other takes a `Subscriber[T] => Unit` function, because expressions like
+  `o => Subscription{}` have both of these types.
+  So we call the old create method "create", and the new create method "apply".
+  Try it out yourself here:
+  def foo[T]: Unit = { 
+    val fMeant: Observer[T] => Subscription = o => Subscription{}
+    val fWrong: Subscriber[T] => Unit = o => Subscription{}
+  }
+  */
+
+  /**
+   * Returns an Observable that will execute the specified function when a someone subscribes to it.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/create.png">
+   *
+   * Write the function you pass so that it behaves as an Observable: It should invoke the
+   * Subscriber's `onNext`, `onError`, and `onCompleted` methods appropriately.
+   *
+   * See <a href="http://go.microsoft.com/fwlink/?LinkID=205219">Rx Design Guidelines (PDF)</a> for detailed
+   * information.
+   *
+   * @tparam T
+   *            the type of the items that this Observable emits
+   * @param f
+   *            a function that accepts a `Subscriber[T]`, and invokes its `onNext`,
+   *            `onError`, and `onCompleted` methods as appropriate
+   * @return an Observable that, when someone subscribes to it, will execute the specified
+   *         function
+   */
+  def apply[T](f: Subscriber[T] => Unit): Observable[T] = {
+    toScalaObservable(rx.Observable.create(f))
+  }
+  
   /**
    * Returns an Observable that invokes an [[rx.lang.scala.Observer]]'s [[rx.lang.scala.Observer.onError onError]]
    * method when the Observer subscribes to it.

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

@@ -56,7 +56,7 @@ trait Observer[-T] {
 
 }
 
-object Observer {
+object Observer extends ObserverFactoryMethods[Observer] {
 
   /**
    * Scala calls XXX; Java receives XXX.
@@ -72,17 +72,25 @@ object Observer {
 
    }
 
-  def apply[T](                                                                ): Observer[T] = apply[T]((v:T)=>(), (e: Throwable)=>(), ()=>())
-  def apply[T](onNext: T=>Unit                                                 ): Observer[T] = apply[T](onNext, (e: Throwable)=>(), ()=>())
-  def apply[T](onNext: T=>Unit, onError: Throwable=>Unit                       ): Observer[T] = apply[T](onNext, onError, ()=>())
-  def apply[T](onNext: T=>Unit,                           onCompleted: ()=>Unit): Observer[T] = apply[T](onNext, (e: Throwable)=>(), onCompleted)
-  def apply[T](onNext: T=>Unit, onError: Throwable=>Unit, onCompleted: ()=>Unit): Observer[T] = {
-      val n = onNext; val e = onError; val c = onCompleted
-      // Java calls XXX; Scala receives XXX.
-      Observer(new rx.Observer[T] {
-         override def onNext(value: T): Unit = n(value)
-         override def onError(error: Throwable): Unit = e(error)
-         override def onCompleted(): Unit = c()
-       })
+  def apply[T](onNext: T => Unit, onError: Throwable => Unit, onCompleted: () => Unit): Observer[T] = {
+    val n = onNext; val e = onError; val c = onCompleted
+    // Java calls XXX; Scala receives XXX.
+    Observer(new rx.Observer[T] {
+      override def onNext(value: T): Unit = n(value)
+      override def onError(error: Throwable): Unit = e(error)
+      override def onCompleted(): Unit = c()
+    })
   }
-}
+}
+
+
+private [scala] trait ObserverFactoryMethods[P[_] <: Observer[_]] {
+
+  def apply[T](onNext: T => Unit, onError: Throwable => Unit, onCompleted: () => Unit): P[T]
+  
+  def apply[T](                                                                ): P[T] = apply[T]((v:T)=>(), (e: Throwable)=>(), ()=>())
+  def apply[T](onNext: T=>Unit                                                 ): P[T] = apply[T](onNext,    (e: Throwable)=>(), ()=>())
+  def apply[T](onNext: T=>Unit, onError: Throwable=>Unit                       ): P[T] = apply[T](onNext,    onError,            ()=>())
+  def apply[T](onNext: T=>Unit,                           onCompleted: ()=>Unit): P[T] = apply[T](onNext,    (e: Throwable)=>(), onCompleted)    
+}
+

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

@@ -0,0 +1,52 @@
+package rx.lang.scala
+
+trait Subscriber[-T] extends Observer[T] with Subscription {
+
+  self =>
+  
+  private [scala] override val asJavaObserver: rx.Observer[_ >: T] = asJavaSubscriber
+  private [scala] override val asJavaSubscription: rx.Subscription = asJavaSubscriber
+  
+  private [scala] val asJavaSubscriber: rx.Subscriber[_ >: T] = new rx.Subscriber[T] {
+    def onNext(value: T): Unit = self.onNext(value)
+    def onError(error: Throwable): Unit = self.onError(error)
+    def onCompleted(): Unit = self.onCompleted()
+  }
+  
+  /**
+   * Used to register an unsubscribe callback.
+   */
+  final def add(s: Subscription): Unit = {
+    asJavaSubscriber.add(s.asJavaSubscription)
+  }
+
+  override final def unsubscribe(): Unit = {
+    asJavaSubscriber.unsubscribe()
+  }
+
+  override final def isUnsubscribed: Boolean = {
+    asJavaSubscriber.isUnsubscribed()
+  }
+
+}
+
+object Subscriber extends ObserverFactoryMethods[Subscriber] {
+
+  private[scala] def apply[T](subscriber: rx.Subscriber[T]): Subscriber[T] = new Subscriber[T] {
+    override val asJavaSubscriber = subscriber
+
+    override def onNext(value: T): Unit = asJavaSubscriber.onNext(value)
+    override def onError(error: Throwable): Unit = asJavaSubscriber.onError(error)
+    override def onCompleted(): Unit = asJavaSubscriber.onCompleted()
+  }
+
+  def apply[T](onNext: T => Unit, onError: Throwable => Unit, onCompleted: () => Unit): Subscriber[T] = {
+    val n = onNext; val e = onError; val c = onCompleted
+    // Java calls XXX; Scala receives XXX.
+    Subscriber(new rx.Subscriber[T] {
+      override def onNext(value: T): Unit = n(value)
+      override def onError(error: Throwable): Unit = e(error)
+      override def onCompleted(): Unit = c()
+    })
+  }
+}