RxScala: Add *Option and *OrElse to BlockingObservable

Author: zsxwing

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

@@ -2862,13 +2862,14 @@ trait Observable[+T]
 
   /**
    * 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`.
+   * item. If the source Observable emits more than one item or no items, notify of an `IllegalArgumentException`
+   * or `NoSuchElementException` respectively.
    * 
    * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/single.png">
    * 
    * @return an Observable that emits the single item emitted by the source Observable
-   * @throws NoSuchElementException
-   *             if the source emits more than one item or no items
+   * @throws IllegalArgumentException if the source emits more than one item
+   * @throws NoSuchElementException if the source emits no items
    * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-single-and-singleordefault">RxJava Uncyclo: single()</a>
    * @see "MSDN: Observable.singleAsync()"
    */
@@ -3093,7 +3094,7 @@ trait Observable[+T]
    */
   @deprecated("Use `toBlocking` instead", "0.19")
   def toBlockingObservable: BlockingObservable[T] = {
-    new BlockingObservable[T](asJavaObservable.toBlocking)
+    new BlockingObservable[T](this)
   }
 
   /**
@@ -3105,7 +3106,7 @@ trait Observable[+T]
    * @since 0.19
    */
   def toBlocking: BlockingObservable[T] = {
-    new BlockingObservable[T](asJavaObservable.toBlocking)
+    new BlockingObservable[T](this)
   }
 
   /**

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/observables/BlockingObservable.scala

@@ -17,18 +17,21 @@ package rx.lang.scala.observables
 
 import scala.collection.JavaConverters._
 import rx.lang.scala.ImplicitFunctionConversions._
+import rx.lang.scala.Observable
+import rx.observables.{BlockingObservable => JBlockingObservable}
 
 
 /**
  * An Observable that provides blocking operators.
  * 
- * You can obtain a BlockingObservable from an Observable using [[rx.lang.scala.Observable.toBlockingObservable]]
+ * You can obtain a BlockingObservable from an Observable using [[rx.lang.scala.Observable.toBlocking]]
  */
-// constructor is private because users should use Observable.toBlockingObservable
-class BlockingObservable[+T] private[scala] (val asJava: rx.observables.BlockingObservable[_ <: T]) 
-  extends AnyVal 
+// constructor is private because users should use Observable.toBlocking
+class BlockingObservable[+T] private[scala] (val o: Observable[T])
+  extends AnyVal
 {
-
+  // This is def because "field definition is not allowed in value class"
+  private def asJava: JBlockingObservable[_ <: T] = o.asJavaObservable.toBlocking
   /**
    * Invoke a method on each item emitted by the {@link Observable}; block until the Observable
    * completes.
@@ -69,6 +72,34 @@ class BlockingObservable[+T] private[scala] (val asJava: rx.observables.Blocking
     asJava.last : T
   }
 
+  /**
+   * Returns an `Option` with the last item emitted by the source Observable,
+   * or `None` if the source Observable completes without emitting any items.
+   *
+   * @return an `Option` with the last item emitted by the source Observable,
+   *         or `None` if the source Observable is empty
+   */
+  def lastOption: Option[T] = {
+    o.lastOption.toBlocking.single
+  }
+
+  /**
+   * Returns 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 the last item emitted by the source Observable, or a default item if the source Observable is empty
+   */
+  def lastOrElse[U >: T](default: => U): U = {
+    lastOption match {
+      case Some(element) => element
+      case None => default
+    }
+  }
+
   /**
    * Returns the first item emitted by a specified [[Observable]], or
    * `NoSuchElementException` if source contains no elements.
@@ -96,12 +127,32 @@ class BlockingObservable[+T] private[scala] (val asJava: rx.observables.Blocking
    */
   def head : T = first
 
-  // last                 -> use toIterable.last
-  // lastOrDefault        -> use toIterable.lastOption
-  // first                -> use toIterable.head
-  // firstOrDefault       -> use toIterable.headOption
-  // single(predicate)    -> use filter and single
-  // singleOrDefault      -> use singleOption
+  /**
+   * Returns an `Option` with the very first item emitted by the source Observable,
+   * or `None` if the source Observable is empty.
+   *
+   * @return an `Option` with the very first item from the source,
+   *         or `None` if the source Observable completes without emitting any item.
+   */
+  def headOption: Option[T] = {
+    o.headOption.toBlocking.single
+  }
+
+  /**
+   * Returns the very first item emitted by the source Observable, or a default value if the source Observable is empty.
+   *
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/firstOrDefault.png">
+   *
+   * @param default The default value to emit if the source Observable doesn't emit anything.
+   *                This is a by-name parameter, so it is only evaluated if the source Observable doesn't emit anything.
+   * @return the very first item from the source, or a default value if the source Observable completes without emitting any item.
+   */
+  def headOrElse[U >: T](default: => U): U = {
+    headOption match {
+      case Some(element) => element
+      case None => default
+    }
+  }
 
   /**
    * Returns an {@link Iterable} that always returns the item most recently emitted by an {@link Observable}.
@@ -130,29 +181,50 @@ class BlockingObservable[+T] private[scala] (val asJava: rx.observables.Blocking
   }
 
   /**
-   * If this {@link Observable} completes after emitting a single item, return that item,
-   * otherwise throw an exception.
-   * <p>
-   * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.single.png">
+   * If the source Observable completes after emitting a single item, return that item. If the source Observable
+   * emits more than one item or no items, notify of an `IllegalArgumentException` or `NoSuchElementException` respectively.
    *
-   * @return the single item emitted by the {@link Observable}
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/single.png">
+   *
+   * @return an Observable that emits the single item emitted by the source Observable
+   * @throws IllegalArgumentException if the source emits more than one item
+   * @throws NoSuchElementException if the source emits no items
    */
   def single: T = {
     asJava.single(): T // useless ascription because of compiler bug
   }
 
   /**
-   * If this {@link Observable} completes after emitting a single item, return an Option containing
-   * this item, otherwise return {@code None}.
+   * If the source Observable completes after emitting a single item, return an `Option` with that item;
+   * if the source Observable is empty, return `None`. If the source Observable emits more than one item,
+   * throw an `IllegalArgumentException`.
+   *
+   * @return 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: Option[T] = {
-    var size: Int = 0
-    var last: Option[T] = None
-    for (t <- toIterable) {
-      size += 1
-      last = Some(t)
+    o.singleOption.toBlocking.single
+  }
+
+  /**
+   * If the source Observable completes after emitting a single item, return that item;
+   * if the source Observable is empty, return 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 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): U = {
+    singleOption match {
+      case Some(element) => element
+      case None => default
     }
-    if (size == 1) last else None
   }
 
   // TODO toFuture()

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/observables/BlockingObservableTest.scala

@@ -0,0 +1,116 @@
+package rx.lang.scala.observables
+
+import org.junit.Assert._
+import org.junit.Test
+import org.scalatest.junit.JUnitSuite
+import scala.language.postfixOps
+import rx.lang.scala.Observable
+
+class BlockingObservableTest extends JUnitSuite {
+
+  @Test
+  def testSingleOption() {
+    val o = Observable.items(1)
+    assertEquals(Some(1), o.toBlocking.singleOption)
+  }
+
+  @Test
+  def testSingleOptionWithEmpty() {
+    val o = Observable.empty
+    assertEquals(None, o.toBlocking.singleOption)
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testSingleOptionWithMultipleItems() {
+    Observable.items(1, 2).toBlocking.singleOption
+  }
+
+  @Test
+  def testSingleOrElse() {
+    val o = Observable.items(1)
+    assertEquals(1, o.toBlocking.singleOrElse(2))
+  }
+
+  @Test
+  def testSingleOrElseWithEmpty() {
+    val o = Observable.empty
+    assertEquals(2, o.toBlocking.singleOrElse(2))
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testSingleOrElseWithMultipleItems() {
+    Observable.items(1, 2).toBlocking.singleOrElse(2)
+  }
+
+  @Test
+  def testHeadOption() {
+    val o = Observable.items(1)
+    assertEquals(Some(1), o.toBlocking.headOption)
+  }
+
+  @Test
+  def testHeadOptionWithEmpty() {
+    val o = Observable.empty
+    assertEquals(None, o.toBlocking.headOption)
+  }
+
+  @Test
+  def testHeadOptionWithMultipleItems() {
+    val o = Observable.items(1, 2)
+    assertEquals(Some(1), o.toBlocking.headOption)
+  }
+
+  @Test
+  def testHeadOrElse() {
+    val o = Observable.items(1)
+    assertEquals(1, o.toBlocking.headOrElse(2))
+  }
+
+  @Test
+  def testHeadOrElseWithEmpty() {
+    val o = Observable.empty
+    assertEquals(2, o.toBlocking.headOrElse(2))
+  }
+
+  @Test
+  def testHeadOrElseWithMultipleItems() {
+    val o = Observable.items(1, 2)
+    assertEquals(1, o.toBlocking.headOrElse(2))
+  }
+
+  @Test
+  def testLastOption() {
+    val o = Observable.items(1)
+    assertEquals(Some(1), o.toBlocking.lastOption)
+  }
+
+  @Test
+  def testLastOptionWithEmpty() {
+    val o = Observable.empty
+    assertEquals(None, o.toBlocking.lastOption)
+  }
+
+  @Test
+  def testLastOptionWithMultipleItems() {
+    val o = Observable.items(1, 2)
+    assertEquals(Some(2), o.toBlocking.lastOption)
+  }
+
+  @Test
+  def testLastOrElse() {
+    val o = Observable.items(1)
+    assertEquals(1, o.toBlocking.lastOrElse(2))
+  }
+
+  @Test
+  def testLastOrElseWithEmpty() {
+    val o = Observable.empty
+    assertEquals(2, o.toBlocking.lastOrElse(2))
+  }
+
+  @Test
+  def testLastOrElseWithMultipleItems() {
+    val o = Observable.items(1, 2)
+    assertEquals(2, o.toBlocking.lastOrElse(3))
+  }
+}