Add lift to rxscala

Author: zsxwing

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

@@ -645,4 +645,60 @@ class RxScalaDemo extends JUnitSuite {
     val o : Observable[String] = List("alice", "bob", "carol").toObservable
     assertEquals(List("alice", "bob", "carol"), o.retry(3).toBlockingObservable.toList)
   }
+
+  @Test def liftExample1(): Unit = {
+    // Add "No. " in front of each item
+    val o = List(1, 2, 3).toObservable.lift {
+      subscriber: Subscriber[String] =>
+        Subscriber(
+          subscriber,
+          v => subscriber.onNext("No. " + v),
+          e => subscriber.onError(e),
+          () => subscriber.onCompleted
+        )
+    }.toBlockingObservable.toList
+    println(o)
+  }
+
+  @Test def liftExample2(): Unit = {
+    val o = Observable {
+      subscriber: Subscriber[Int] => {
+        for (i <- 1 to 10 if !subscriber.isUnsubscribed) {
+          println("emit " + i)
+          subscriber.onNext(i)
+        }
+        if (!subscriber.isUnsubscribed) {
+          println("emit onCompleted")
+          subscriber.onCompleted
+        }
+      }
+    }
+    // Take the first 5 items
+    val take = 5
+    val result = o.lift {
+      subscriber: Subscriber[String] =>
+        var index = 0
+        Subscriber(
+          subscriber,
+          v => {
+            if (index < take) {
+              subscriber.onNext("No. " + v)
+            }
+            if (index == take - 1) {
+              subscriber.onCompleted
+            }
+            index += 1
+          },
+          e => subscriber.onError(e),
+          () => subscriber.onCompleted
+        )
+    }.toBlockingObservable.toList
+    println(result)
+    // emit 1
+    // emit 2
+    // emit 3
+    // emit 4
+    // emit 5
+    // List(No. 1, No. 2, No. 3, No. 4, No. 5)
+  }
 }

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

@@ -2537,6 +2537,29 @@ trait Observable[+T]
     toScalaObservable[util.Map[K, V]](o).map(m => mapFactory() ++ m.toMap)
   }
 
+  /**
+   * Lift a function to the current Observable and return a new Observable that when subscribed to will pass
+   * the values of the current Observable through the function.
+   * <p>
+   * In other words, this allows chaining Observers together on an Observable for acting on the values within
+   * the Observable.
+   * {{{
+   * observable.map(...).filter(...).take(5).lift(new ObserverA()).lift(new ObserverB(...)).subscribe()
+   * }}}
+   *
+   * @param operator
+   * @return an Observable that emits values that are the result of applying the bind function to the values
+   *         of the current Observable
+   */
+  def lift[R](operator: Subscriber[_ >: R] => Subscriber[_ >: T]): Observable[R] = {
+    val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val thatJava = thisJava.lift[R](new rx.Observable.Operator[R, T] {
+      override def call(subscriber: rx.Subscriber[_ >: R]): rx.Subscriber[_ >: T] = {
+        toJavaSubscriber(operator.call(toScalaSubscriber(subscriber)))
+      }
+    })
+    toScalaObservable(thatJava)
+  }
 }
 
 /**

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

@@ -49,4 +49,14 @@ object Subscriber extends ObserverFactoryMethods[Subscriber] {
       override def onCompleted(): Unit = c()
     })
   }
+
+  def apply[T](subscriber: Subscriber[_], 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](subscriber.asJavaSubscriber) {
+      override def onNext(value: T): Unit = n(value)
+      override def onError(error: Throwable): Unit = e(error)
+      override def onCompleted(): Unit = c()
+    })
+  }
 }