Add Operator conversion, tail and update liftExample2

Author: zsxwing

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

@@ -650,9 +650,9 @@ class RxScalaDemo extends JUnitSuite {
     // Add "No. " in front of each item
     val o = List(1, 2, 3).toObservable.lift {
       subscriber: Subscriber[String] =>
-        Subscriber(
+        Subscriber[Int](
           subscriber,
-          v => subscriber.onNext("No. " + v),
+          (v: Int)  => subscriber.onNext("No. " + v),
           e => subscriber.onError(e),
           () => subscriber.onCompleted
         )
@@ -661,44 +661,51 @@ class RxScalaDemo extends JUnitSuite {
   }
 
   @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
-        }
-      }
+    // Split the input Strings with " "
+    val splitStringsWithSpace = (subscriber: Subscriber[String]) => {
+      Subscriber[String](
+        subscriber,
+        (v: String) => v.split(" ").foreach(subscriber.onNext(_)),
+        e => subscriber.onError(e),
+        () => subscriber.onCompleted
+      )
+    }
+
+    // Convert the input Strings to Chars
+    val stringsToChars = (subscriber: Subscriber[Char]) => {
+      Subscriber[String](
+        subscriber,
+        (v: String) => v.foreach(subscriber.onNext(_)),
+        e => subscriber.onError(e),
+        () => subscriber.onCompleted
+      )
+    }
+
+    // Skip the first n items. If the length of source is less than n, throw an IllegalArgumentException
+    def skipWithException[T](n: Int) = (subscriber: Subscriber[T]) => {
+      var count = 0
+      Subscriber[T](
+        subscriber,
+        (v: T) => {
+          if (count >= n) subscriber.onNext(v)
+          count += 1
+        },
+        e => subscriber.onError(e),
+        () => if (count < n) subscriber.onError(new IllegalArgumentException("There is no enough items")) else subscriber.onCompleted
+      )
+    }
+
+    val o = List("RxJava – Reactive Extensions for the JVM").toObservable
+      .lift(splitStringsWithSpace)
+      .map(_.toLowerCase)
+      .lift(stringsToChars)
+      .filter(_.isLetter)
+      .lift(skipWithException(100))
+    try {
+      o.toBlockingObservable.toList
+    }
+    catch {
+      case e: IllegalArgumentException => println("IllegalArgumentException from skipWithException")
     }
-    // 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/JavaConversions.scala

@@ -54,4 +54,12 @@ object JavaConversions {
       val asJavaObservable = observable
     }
   }
+
+  implicit def toJavaOperator[T, R](operator: Subscriber[R] => Subscriber[_ >: T]): rx.Observable.Operator[R, T] = {
+    new rx.Observable.Operator[R, T] {
+      override def call(subscriber: rx.Subscriber[_ >: R]): rx.Subscriber[_ >: T] = {
+        toJavaSubscriber[T](operator(toScalaSubscriber[R](subscriber)))
+      }
+    }
+  }
 }

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

@@ -2057,7 +2057,28 @@ trait Observable[+T]
    * @see [[Observable.first]]
    */
   def head: Observable[T] = first
-  
+
+  /**
+   * Returns an Observable that emits all items except the first one, or raises an `UnsupportedOperationException`
+   * if the source Observable is empty.
+   *
+   * @return an Observable that emits all items except the first one, or raises an `UnsupportedOperationException`
+   *         if the source Observable is empty.
+   */
+  def tail: Observable[T] = {
+    lift {
+      (subscriber: Subscriber[T]) => {
+        var isFirst = true
+        Subscriber[T](
+          subscriber,
+          (v: T) => if(isFirst) isFirst = false else subscriber.onNext(v),
+          e => subscriber.onError(e),
+          () => if(isFirst) subscriber.onError(new UnsupportedOperationException("tail of empty Observable")) else subscriber.onCompleted
+        )
+      }
+    }
+  }
+
   /**
    * Returns an Observable that emits the last item emitted by the source Observable or notifies observers of
    * an `NoSuchElementException` if the source Observable is empty.
@@ -2551,14 +2572,8 @@ trait Observable[+T]
    * @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)
+  def lift[R](operator: Subscriber[R] => Subscriber[_ >: T]): Observable[R] = {
+    toScalaObservable(asJavaObservable.lift(toJavaOperator[T, R](operator)))
   }
 }
 

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

@@ -146,16 +146,26 @@ class ObservableTests extends JUnitSuite {
     assertEquals(l2, l1)
   }
 
-  /*
- @Test def testHead() {
-   val observer = mock(classOf[Observer[Int]])
-   val o = Observable().head
-   val sub = o.subscribe(observer)
-
-   verify(observer, never).onNext(any(classOf[Int]))
-   verify(observer, never).onCompleted()
-   verify(observer, times(1)).onError(any(classOf[NoSuchElementException]))
- }
- */
+  @Test def testHead() {
+    val o: Observable[String] = List("alice", "bob", "carol").toObservable.head
+    assertEquals(List("alice"), o.toBlockingObservable.toList)
+  }
+
+  @Test(expected = classOf[NoSuchElementException])
+  def testHeadWithEmptyObservable() {
+    val o: Observable[String] = List[String]().toObservable.head
+    o.toBlockingObservable.toList
+  }
 
+  @Test def testTail() {
+    val o: Observable[String] = List("alice", "bob", "carol").toObservable.tail
+    assertEquals(List("bob", "carol"), o.toBlockingObservable.toList)
+    assertEquals(List("bob", "carol"), o.toBlockingObservable.toList)
+  }
+
+  @Test(expected = classOf[UnsupportedOperationException])
+  def testTailWithEmptyObservable() {
+    val o: Observable[String] = List[String]().toObservable.tail
+    o.toBlockingObservable.toList
+  }
 }