Add singleOption, singleOrElse, headOption, lastOption, lastOrElse to RxScala

Author: zsxwing

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

@@ -2651,10 +2651,20 @@ trait Observable[+T]
    *         if the source Observable completes without emitting any item.
    */
   def firstOrElse[U >: T](default: => U): Observable[U] = {
-    this.take(1).foldLeft[Option[U]](None)((v: Option[U], e: U) => Some(e)).map({
-      case Some(element) => element
-      case None => default
-    })
+    take(1).singleOrElse(default)
+  }
+
+  /**
+   * Returns an Observable that emits only an `Option` with the very first item emitted by the source Observable,
+   * or `None` if the source Observable is empty.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/firstOrDefault.png">
+   *
+   * @return an Observable that emits only an `Option` with the very first item from the source, or `None`
+   *         if the source Observable completes without emitting any item.
+   */
+  def headOption: Observable[Option[T]] = {
+    take(1).singleOption
   }
 
   /**
@@ -2736,6 +2746,34 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.last)
   }
 
+  /**
+   * Returns an Observable that emits only an `Option` with the last item emitted by the source Observable,
+   * or `None` if the source Observable completes without emitting any items.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/lastOrDefault.png">
+   *
+   * @return an Observable that emits only an `Option` with the last item emitted by the source Observable,
+   *         or `None` if the source Observable is empty
+   */
+  def lastOption: Observable[Option[T]] = {
+    takeRight(1).singleOption
+  }
+
+  /**
+   * Returns an Observable that emits only the last item emitted by the source Observable, or a default item
+   * if the source Observable completes without emitting any items.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/lastOrDefault.png">
+   *
+   * @param default the default item to emit if the source Observable is empty.
+   *                This is a by-name parameter, so it is only evaluated if the source Observable doesn't emit anything.
+   * @return an Observable that emits only the last item emitted by the source Observable, or a default item
+   *         if the source Observable is empty
+   */
+  def lastOrElse[U >: T](default: => U): Observable[U] = {
+    takeRight(1).singleOrElse(default)
+  }
+
   /**
    * If the source Observable completes after emitting a single item, return an Observable that emits that
    * item. If the source Observable emits more than one item or no items, throw an `NoSuchElementException`.
@@ -2752,6 +2790,42 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.single)
   }
 
+  /**
+   * If the source Observable completes after emitting a single item, return an Observable that emits an `Option`
+   * with that item; if the source Observable is empty, return an Observable that emits `None`.
+   * If the source Observable emits more than one item, throw an `IllegalArgumentException`.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/singleOrDefault.png">
+   *
+   * @return an Observable that emits an `Option` with the single item emitted by the source Observable, or
+   *         `None` if the source Observable is empty
+   * @throws IllegalArgumentException if the source Observable emits more than one item
+   */
+  def singleOption: Observable[Option[T]] = {
+    val jObservableOption = map(Some(_)).asJavaObservable.asInstanceOf[rx.Observable[Option[T]]]
+    toScalaObservable[Option[T]](jObservableOption.singleOrDefault(None))
+  }
+
+  /**
+   * If the source Observable completes after emitting a single item, return an Observable that emits that
+   * item; if the source Observable is empty, return an Observable that emits a default item. If the source
+   * Observable emits more than one item, throw an `IllegalArgumentException`.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/singleOrDefault.png">
+   *
+   * @param default a default value to emit if the source Observable emits no item.
+   *                This is a by-name parameter, so it is only evaluated if the source Observable doesn't emit anything.
+   * @return an Observable that emits the single item emitted by the source Observable, or a default item if
+   *         the source Observable is empty
+   * @throws IllegalArgumentException if the source Observable emits more than one item
+   */
+  def singleOrElse[U >: T](default: => U): Observable[U] = {
+    singleOption.map {
+      case Some(element) => element
+      case None => default
+    }
+  }
+
   /**
    * Returns an Observable that forwards all sequentially distinct items emitted from the source Observable.
    *

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

@@ -89,6 +89,9 @@ class CompletenessTest extends JUnitSuite {
       "groupBy(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: R])" -> "[use `groupBy` and `map`]",
       "groupByUntil(Func1[_ >: T, _ <: TKey], Func1[_ >: GroupedObservable[TKey, T], _ <: Observable[_ <: TDuration]])" -> "groupByUntil(T => K, (K, Observable[T]) => Observable[Any])",
       "ignoreElements()" -> "[use `filter(_ => false)`]",
+      "last(Func1[_ >: T, Boolean])" -> "[use `filter(predicate).last`]",
+      "lastOrDefault(T)" -> "lastOrElse(=> U)",
+      "lastOrDefault(T, Func1[_ >: T, Boolean])" -> "[use `filter(predicate).lastOrElse(default)`]",
       "lift(Operator[_ <: R, _ >: T])" -> "lift(Subscriber[R] => Subscriber[T])",
       "limit(Int)" -> "take(Int)",
       "mapWithIndex(Func2[_ >: T, Integer, _ <: R])" -> "[combine `zipWithIndex` with `map` or with a for comprehension]",
@@ -117,13 +120,15 @@ class CompletenessTest extends JUnitSuite {
       "sample(Observable[U])" -> "sample(Observable[Any])",
       "scan(Func2[T, T, T])" -> unnecessary,
       "scan(R, Func2[R, _ >: T, R])" -> "scan(R)((R, T) => R)",
+      "single(Func1[_ >: T, Boolean])" -> "[use `filter(predicate).single`]",
+      "singleOrDefault(T)" -> "singleOrElse(=> U)",
+      "singleOrDefault(T, Func1[_ >: T, Boolean])" -> "[use `filter(predicate).singleOrElse(default)`]",
       "skip(Int)" -> "drop(Int)",
       "skip(Long, TimeUnit)" -> "drop(Duration)",
       "skip(Long, TimeUnit, Scheduler)" -> "drop(Duration, Scheduler)",
       "skipWhile(Func1[_ >: T, Boolean])" -> "dropWhile(T => Boolean)",
       "skipWhileWithIndex(Func2[_ >: T, Integer, Boolean])" -> unnecessary,
       "skipUntil(Observable[U])" -> "dropUntil(Observable[E])",
-      "single(Func1[_ >: T, Boolean])" -> "[use `filter(predicate).single`]",
       "startWith(T)" -> "[use `item +: o`]",
       "startWith(Array[T])" -> "[use `Observable.items(items) ++ o`]",
       "startWith(Array[T], Scheduler)" -> "[use `Observable.items(items).subscribeOn(scheduler) ++ o`]",
@@ -159,8 +164,6 @@ class CompletenessTest extends JUnitSuite {
       "window(Observable[_ <: TOpening], Func1[_ >: TOpening, _ <: Observable[_ <: TClosing]])" -> "window(Observable[Opening], Opening => Observable[Any])",
       "window(Long, Long, TimeUnit)" -> "window(Duration, Duration)",
       "window(Long, Long, TimeUnit, Scheduler)" -> "window(Duration, Duration, Scheduler)",
-      "zip(Observable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zipWith(Observable[U], (T, U) => R)",
-      "zip(Iterable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zipWith(Iterable[U], (T, U) => R)",
 
       // manually added entries for Java static methods
       "average(Observable[Integer])" -> averageProblem,
@@ -197,7 +200,9 @@ class CompletenessTest extends JUnitSuite {
       "switchOnNext(Observable[_ <: Observable[_ <: T]])" -> "switch(<:<[Observable[T], Observable[Observable[U]]])",
       "zip(Observable[_ <: T1], Observable[_ <: T2], Func2[_ >: T1, _ >: T2, _ <: R])" -> "[use instance method `zip` and `map`]",
       "zip(Observable[_ <: Observable[_]], FuncN[_ <: R])" -> "[use `zip` in companion object and `map`]",
-      "zip(Iterable[_ <: Observable[_]], FuncN[_ <: R])" -> "[use `zip` in companion object and `map`]"
+      "zip(Iterable[_ <: Observable[_]], FuncN[_ <: R])" -> "[use `zip` in companion object and `map`]",
+      "zip(Observable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zipWith(Observable[U], (T, U) => R)",
+      "zip(Iterable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zipWith(Iterable[U], (T, U) => R)"
   ) ++ List.iterate("T, T", 8)(s => s + ", T").map(
       // all 9 overloads of startWith:
       "startWith(" + _ + ")" -> "[use `Observable.items(...) ++ o`]"
@@ -345,6 +350,7 @@ class CompletenessTest extends JUnitSuite {
     println(  "----------------------------------------------\n")
 
     val actualMethods = getPublicInstanceAndCompanionMethods(typeOf[rx.lang.scala.Observable[_]]).toSet
+    actualMethods.toList.sorted.foreach(println)
     var good = 0
     var bad = 0
     for ((javaM, scalaM) <- SortedMap(correspondence.toSeq :_*)) {

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

@@ -176,4 +176,44 @@ class ObservableTests extends JUnitSuite {
     assertEquals(List(0, 1, 2), o.toBlockingObservable.toList)
   }
 
+  @Test
+  def testSingleOrElse() {
+    val o = Observable.items(1).singleOrElse(2)
+    assertEquals(1, o.toBlocking.single)
+  }
+
+  @Test
+  def testSingleOrElseWithEmptyObservable() {
+    val o: Observable[Int] = Observable.empty.singleOrElse(1)
+    assertEquals(1, o.toBlocking.single)
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testSingleOrElseWithTooManyItems() {
+    Observable.items(1, 2).singleOrElse(1).toBlocking.single
+  }
+
+  @Test
+  def testSingleOrElseWithCallByName() {
+    var called = false
+    val o: Observable[Int] = Observable.empty.singleOrElse {
+      called = true
+      1
+    }
+    assertFalse(called)
+    o.subscribe()
+    assertTrue(called)
+  }
+
+  @Test
+  def testSingleOrElseWithCallByName2() {
+    var called = false
+    val o = Observable.items(1).singleOrElse {
+      called = true
+      2
+    }
+    assertFalse(called)
+    o.subscribe()
+    assertFalse(called)
+  }
 }