Double negation Observer.

Author: AppliedDuality

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

@@ -22,8 +22,6 @@ import rx.Observable.OnSubscribeFunc
 /**
  * The Observable interface that implements the Reactive Pattern.
  *
- * @param asJavaObservable the underlying Java observable
- *
  * @define subscribeObserverMain
  * Call this method to subscribe an [[rx.lang.scala.Observer]] for receiving 
  * items and notifications from the Observable.
@@ -1855,7 +1853,7 @@ object Observable {
   /**
    * Creates a new Scala Observable from a given Java Observable.
    */
-  def apply[T](observable: rx.Observable[_ <: T]): Observable[T] = {
+  private [scala] def apply[T](observable: rx.Observable[_ <: T]): Observable[T] = {
     new Observable[T]{
       def asJavaObservable = observable
     }
@@ -1878,13 +1876,13 @@ object Observable {
    *
    *
    * @tparam T
-   *            the type of the items that this Observable emits
+   *            the type of the items that this Observable emits.
    * @param func
    *            a function that accepts an `Observer[T]`, invokes its `onNext`, `onError`, and `onCompleted` methods
    *            as appropriate, and returns a [[rx.lang.scala.Subscription]] to allow the Observer to
-   *            canceling the subscription
-   * @return an Observable that, when an [[rx.lang.scala.Observer]] subscribes to it, will execute the given
-   *         function
+   *            canceling the subscription.
+   * @return
+   *         an Observable that, when an [[rx.lang.scala.Observer]] subscribes to it, will execute the given function.
    */
   def apply[T](func: Observer[T] => Subscription): Observable[T] = {
     Observable[T](rx.Observable.create(new OnSubscribeFunc[T] {
@@ -1903,15 +1901,17 @@ object Observable {
   }
 
   /**
-   * Returns an Observable that invokes an [[rx.lang.scala.Observer]]'s [[rx.lang.scala.Observer.onError onError]] method when the Observer subscribes to it
+   * Returns an Observable that invokes an [[rx.lang.scala.Observer]]'s [[rx.lang.scala.Observer.onError onError]]
+   * method when the Observer subscribes to it.
    *
    * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/error.png">
    *
    * @param exception
    *            the particular error to report
    * @tparam T
    *            the type of the items (ostensibly) emitted by the Observable
-   * @return an Observable that invokes the [[rx.lang.scala.Observer]]'s [[rx.lang.scala.Observer.onError onError]] method when the Observer subscribes to it
+   * @return an Observable that invokes the [[rx.lang.scala.Observer]]'s [[rx.lang.scala.Observer.onError onError]]
+   *         method when the Observer subscribes to it
    */
   def apply[T](exception: Throwable): Observable[T] = {
     Observable[T](rx.Observable.error(exception))

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

@@ -26,7 +26,11 @@ import rx.joins.ObserverBase
 */
 trait Observer[-T] {
 
-  def asJavaObserver: rx.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)
+  }
 
  /**
  * Provides the Observer with new data.
@@ -35,78 +39,35 @@ 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 = asJavaObserver.onNext(value)
+  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 = asJavaObserver.onError(error)
+  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  = asJavaObserver.onCompleted()
+  def onCompleted(): Unit
 
 }
 
 object Observer {
-
-  private [scala] def apply[T](observer: rx.Observer[T]) : Observer[T] = {
-    new Observer[T]() {
-      def asJavaObserver: rx.Observer[_ >: T] = observer
-    }
-  }
-
   /**
-   * Creates an [[rx.lang.scala.Observer]]
-   * @param onNext the onNext action
-   * @param onError the onError action
-   * @param onCompleted the onCompleted action
+   * Assume that the underlying rx.Observer does not need to be wrapped
    */
-  def apply[T](onNext: T => Unit, onError: Throwable => Unit, onCompleted: () => Unit): Observer[T] = {
-    Observer(new ObserverBase[T] () {
-      protected def onCompletedCore(): Unit = onCompleted()
-      protected def onErrorCore(error: Throwable): Unit = onError(error)
-      protected def onNextCore(value: T): Unit = onNext(value)
-    })
-  }
-
-  /**
-   * Creates an [[rx.lang.scala.Observer]]
-   * @param onNext the onNext action
-   * @param onError the onError action
-   */
-  def apply[T](onNext: T => Unit, onError: Throwable => Unit): Observer[T] = {
-    Observer(new ObserverBase[T] () {
-      protected def onCompletedCore(): Unit = {}
-      protected def onErrorCore(error: Throwable): Unit = onError(error)
-      protected def onNextCore(value: T): Unit = onNext(value)
-    })
-  }
-
-  def apply[T](onNext: T => Unit, onCompleted: () => Unit): Observer[T] = {
-    Observer(new ObserverBase[T] () {
-      protected def onCompletedCore(): Unit = onCompleted()
-      protected def onErrorCore(error: Throwable): Unit = {}
-      protected def onNextCore(value: T): Unit = onNext(value)
-    })
-  }
-
-  /**
-   * Creates an [[rx.lang.scala.Observer]]
-   * @param onNext the onNext action
-   */
-  def apply[T](onNext: T => Unit): Observer[T] = {
-    Observer(new ObserverBase[T] () {
-      protected def onCompletedCore(): Unit = {}
-      protected def onErrorCore(error: Throwable): Unit = {}
-      protected def onNextCore(value: T): Unit = onNext(value)
-    })
-  }
-}
-
+  private [scala] def apply[T](observer: rx.Observer[T]) : Observer[T] = {
+     new Observer[T]() {
+       override def asJavaObserver: rx.Observer[_ >: T] = observer
 
+       def onCompleted(): Unit = asJavaObserver.onCompleted()
+       def onError(error: Throwable): Unit = asJavaObserver.onError(error)
+       def onNext(value: T): Unit = asJavaObserver.onNext(value)
+     }
+   }
+}

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

@@ -21,6 +21,10 @@ package rx.lang.scala
 trait Subject[-T, +R] extends Observable[R] with Observer[T] {
   val asJavaSubject:    rx.subjects.Subject[_ >: T, _<: R]
   def asJavaObservable: rx.Observable[_ <: R] = asJavaSubject
-  def asJavaObserver:   rx.Observer[_ >: T] = asJavaSubject
+  override def asJavaObserver:   rx.Observer[_ >: T] = asJavaSubject
+
+  def onCompleted(): Unit = asJavaObserver.onCompleted()
+  def onError(error: Throwable): Unit = asJavaObserver.onError(error)
+  def onNext(value: T): Unit = asJavaObserver.onNext(value)
 }