Subjects & Observers

Author: AppliedDuality

language-adaptors/rxjava-scala/ReleaseNotes.md

@@ -34,7 +34,7 @@ and implement any of the methods that you care about:
    }
 ```
  or you can use one of the overloads of the companion `Observer` object by passing in implementations of the `onNext`,
- `onError` or `onCompleted` methods. The advantage of this is that you get type inference as in `Observer(println(_))`.
+ `onError` or `onCompleted` methods.
 
 Note that typically you do not need to create an `Observer` since all of the methods that accept an `Observer[T]`
 (for instance `subscribe`) usually come with overloads that accept the individual methods

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

@@ -79,7 +79,7 @@ trait Observable[+T]
   import rx.lang.scala.observables.BlockingObservable
   import rx.lang.scala.ImplicitFunctionConversions._
 
-  private [scala] def asJavaObservable: rx.Observable[_ <: T]
+  private [scala] val asJavaObservable: rx.Observable[_ <: T]
 
   /**
    * $subscribeObserverMain

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

@@ -16,6 +16,7 @@
 package rx.lang.scala
 
 import rx.joins.ObserverBase
+import java.util.concurrent.atomic.AtomicBoolean
 
 /**
  Provides a mechanism for receiving push-based notifications.
@@ -26,11 +27,7 @@ import rx.joins.ObserverBase
 */
 trait Observer[-T] {
 
-  private [scala] def asJavaObserver: rx.Observer[_ >: T] = new ObserverBase[T] {
-    protected def onCompletedCore(): Unit = onCompleted()
-    protected def onErrorCore(error: Throwable): Unit = onError(error)
-    protected def onNextCore(value: T): Unit = onNext(value)
-  }
+  private [scala] val asJavaObserver: rx.Observer[_ >: T]
 
  /**
  * Provides the Observer with new data.
@@ -39,50 +36,51 @@ trait Observer[-T] {
  *
  * The [[rx.lang.scala.Observable]] will not call this method again after it calls either `onCompleted` or `onError`.
  */
-  def onNext(value: T): Unit = {}
+  def onNext(value: T): Unit
 
   /**
   * Notifies the Observer that the [[rx.lang.scala.Observable]] has experienced an error condition.
   *
   * If the [[rx.lang.scala.Observable]] calls this method, it will not thereafter call `onNext` or `onCompleted`.
   */
-  def onError(error: Throwable): Unit = {}
+  def onError(error: Throwable): Unit
 
   /**
    * Notifies the Observer that the [[rx.lang.scala.Observable]] has finished sending push-based notifications.
    *
    * The [[rx.lang.scala.Observable]] will not call this method if it calls `onError`.
    */
-  def onCompleted(): Unit = {}
+  def onCompleted(): Unit
 
 }
 
 object Observer {
+
   /**
-   * Assume that the underlying rx.Observer does not need to be wrapped.
+   * Scala calls XXX; Java receives XXX.
    */
   private [scala] def apply[T](observer: rx.Observer[T]) : Observer[T] = {
      new Observer[T] {
 
-       override def asJavaObserver = observer
-
-       override def onNext(value: T): Unit = asJavaObserver.onNext(value)
-       override def onError(error: Throwable): Unit = asJavaObserver.onError(error)
-       override def onCompleted(): Unit = asJavaObserver.onCompleted()
+       val asJavaObserver = observer
 
+       def onNext(value: T): Unit = asJavaObserver.onNext(value)
+       def onError(error: Throwable): Unit = asJavaObserver.onError(error)
+       def onCompleted(): Unit = asJavaObserver.onCompleted()
      }
+
    }
 
   def apply[T](                                                                ): Observer[T] = apply(v=>{}, e=>{}, ()=>{})
   def apply[T](onNext: T=>Unit                                                 ): Observer[T] = apply(onNext, e=>{}, ()=>{})
   def apply[T](onNext: T=>Unit, onError: Throwable=>Unit                       ): Observer[T] = apply(onNext, onError, ()=>{})
   def apply[T](onNext: T=>Unit,                           onCompleted: ()=>Unit): Observer[T] = apply(onNext, e=>{}, onCompleted)
-  def apply[T](onNext: T=>Unit, onError: Throwable=>Unit, onCompleted: ()=>Unit): Observer[T] = {
-       val n = onNext; val e = onError; val c = onCompleted
-       new 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](n: T=>Unit, e: Throwable=>Unit, c: ()=>Unit): Observer[T] = {
+      // Java calls XXX; Scala receives XXX.
+      Observer(new rx.Observer[T]{
+         def onNext(value: T): Unit = n(value)
+         def onError(error: Throwable): Unit = e(error)
+         def onCompleted(): Unit = c()
+       })
   }
 }

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

@@ -23,18 +23,13 @@ import rx.joins.ObserverBase
 trait Subject[-T, +R] extends Observable[R] with Observer[T] {
   private [scala] val asJavaSubject: rx.subjects.Subject[_ >: T, _<: R]
 
-  def asJavaObservable: rx.Observable[_ <: R] = asJavaSubject
+  val asJavaObservable: rx.Observable[_ <: R] = asJavaSubject
 
-  // temporary hack to workaround bugs in rx Subjects
-  override def asJavaObserver: rx.Observer[_ >: T] = new ObserverBase[T] {
-    protected def onNextCore(value: T) = asJavaSubject.onNext(value)
-    protected def onErrorCore(error: Throwable) =  asJavaSubject.onError(error)
-    protected def onCompletedCore() = asJavaSubject.onCompleted()
-  }
+  val asJavaObserver: rx.Observer[_ >: T] = asJavaSubject
+  def onNext(value: T): Unit = { asJavaObserver.onNext(value)}
+  def onError(error: Throwable): Unit = { asJavaObserver.onError(error)  }
+  def onCompleted() { asJavaObserver.onCompleted() }
 
-  override def onNext(value: T): Unit = asJavaObserver.onNext(value)
-  override def onError(error: Throwable): Unit = asJavaObserver.onError(error)
-  override def onCompleted(): Unit = asJavaObserver.onCompleted()
 
 }
 

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

@@ -1,58 +1,95 @@
 package rx.lang.scala
 
+import org.junit.{Assert, Test}
+import org.scalatest.junit.JUnitSuite
+import scala.concurrent.duration._
+import scala.language.postfixOps
+import rx.lang.scala.schedulers.TestScheduler
+import rx.lang.scala.subjects.BehaviorSubject
+import org.mockito.Mockito._
+import org.mockito.Matchers._
 
-//package rx.lang.scala.examples
-//
-//import org.junit.{Assert, Test}
-//import org.scalatest.junit.JUnitSuite
-//import scala.concurrent.duration._
-//import scala.language.postfixOps
-//import rx.lang.scala.{ Observable, Observer }
-//import rx.lang.scala.schedulers.TestScheduler
-//import rx.lang.scala.subjects.BehaviorSubject
-//import org.mockito.Mockito._
-//import org.mockito.Matchers._
-//
-//  @Test def PublishSubjectIsAChannel() {
-//
-//    val channel: BehaviorSubject[Integer] = BehaviorSubject(2013)
-//    val observerA: Observer[Integer]  = mock(classOf[Observer[Integer]])
-//    val observerB: Observer[Integer] = mock(classOf[Observer[Integer]])
-//    val observerC: Observer[Integer] = mock(classOf[Observer[Integer]])
-//
-//    val x = inOrder(observerA, observerB, observerC)
-//
-//    val a = channel.subscribe(observerA)
-//    val b = channel.subscribe(observerB)
-//
-//    x.verify(observerA).onNext(2013)
-//    x.verify(observerB).onNext(2013)
-//
-//    channel.onNext(42)
-//
-//    x.verify(observerA).onNext(42)
-//    x.verify(observerB).onNext(42)
-//
-//    a.unsubscribe()
-//
-//    channel.onNext(4711)
-//
-//    x.verify(observerA, never()).onNext(any())
-//    x.verify(observerB).onNext(4711)
-//
-//    channel.onCompleted()
-//
-//    x.verify(observerA, never()).onCompleted()
-//    x.verify(observerB).onCompleted()
-//
-//    val c = channel.subscribe(observerC)
-//
-//    x.verify(observerC).onCompleted()
-//
-//    channel.onNext(13)
-//
-//    x.verifyNoMoreInteractions()
-//
-//  }
-//
-//}
+
+/**
+ * No fucking clue how to properly mock traits.
+ * Some old-school imperative code works just as well.
+ */
+class SubjectTest extends JUnitSuite {
+
+  @Test def PublishSubjectIsAChannel() {
+
+    var lastA: Integer = null
+    var errorA: Throwable = null
+    var completedA: Boolean = false
+    val observerA = Observer[Integer](
+      (next: Integer) => { lastA = next },
+      (error: Throwable) => { errorA = error },
+      () => { completedA = true }
+    )
+
+    var lastB: Integer = null
+    var errorB: Throwable = null
+    var completedB: Boolean = false
+    val observerB = Observer[Integer](
+      (next: Integer) => { lastB = next },
+      (error: Throwable) => { errorB = error },
+      () => { completedB = true }
+    )
+
+    var lastC: Integer = null
+    var errorC: Throwable = null
+    var completedC: Boolean = false
+    val observerC = Observer[Integer](
+      (next: Integer) => { lastC = next },
+      (error: Throwable) => { errorC = error },
+      () => { completedC = true }
+    )
+
+    val channel: BehaviorSubject[Integer] = BehaviorSubject(2013)
+
+    val a = channel.subscribe(observerA)
+    Assert.assertEquals(2013, lastA)
+
+    val b = channel.subscribe(observerB)
+    Assert.assertEquals(2013, lastB)
+
+    channel.onNext(42)
+    Assert.assertEquals(42, lastA)
+    Assert.assertEquals(42, lastB)
+
+    a.unsubscribe()
+    channel.onNext(4711)
+    Assert.assertEquals(42, lastA)
+    Assert.assertEquals(4711, lastB)
+
+    channel.onCompleted()
+    Assert.assertFalse(completedA)
+    Assert.assertTrue(completedB)
+    Assert.assertEquals(42, lastA)
+    Assert.assertEquals(4711, lastB)
+
+    val c = channel.subscribe(observerC)
+    channel.onNext(13)
+
+    Assert.assertEquals(null, lastC)
+    Assert.assertTrue(completedC)
+
+    Assert.assertFalse(completedA)
+    Assert.assertTrue(completedB)
+    Assert.assertEquals(42, lastA)
+    Assert.assertEquals(4711, lastB)
+
+    channel.onError(new Exception("!"))
+
+    Assert.assertEquals(null, lastC)
+    Assert.assertTrue(completedC)
+
+    Assert.assertFalse(completedA)
+    Assert.assertTrue(completedB)
+    Assert.assertEquals(42, lastA)
+    Assert.assertEquals(4711, lastB)
+
+
+  }
+
+}