Add first, last, single to rxjava-scala

Author: zsxwing

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

@@ -388,6 +388,18 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(10, List(-1, 0, 1).toObservable.filter(condition).firstOrElse(10).toBlockingObservable.single)
   }
 
+  @Test def firstLastSingleExample() {
+    assertEquals(1, List(1, 2, 3, 4).toObservable.head.toBlockingObservable.single)
+    assertEquals(1, List(1, 2, 3, 4).toObservable.first.toBlockingObservable.single)
+    assertEquals(4, List(1, 2, 3, 4).toObservable.last.toBlockingObservable.single)
+    assertEquals(1, List(1).toObservable.single.toBlockingObservable.single)
+
+    assertEquals(1, List(1, 2, 3, 4).toObservable.toBlockingObservable.head)
+    assertEquals(1, List(1, 2, 3, 4).toObservable.toBlockingObservable.first)
+    assertEquals(4, List(1, 2, 3, 4).toObservable.toBlockingObservable.last)
+    assertEquals(1, List(1).toObservable.toBlockingObservable.single)
+  }
+
   def square(x: Int): Int = {
     println(s"$x*$x is being calculated on thread ${Thread.currentThread().getId}")
     Thread.sleep(100) // calculating a square is heavy work :)

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

@@ -1949,36 +1949,64 @@ trait Observable[+T]
   def headOrElse[U >: T](default: => U): Observable[U] = firstOrElse(default)
 
   /**
-   * Returns an Observable that emits only the very first item emitted by the source Observable.
-   * This is just a shorthand for `take(1)`.
-   *
+   * Returns an Observable that emits only the very first item emitted by the source Observable, or raises an
+   * `IllegalArgumentException` if the source Observable is empty.
+   * <p>
    * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/first.png">
-   *
-   * @return an Observable that emits only the very first item from the source, or none if the
-   *         source Observable completes without emitting a single item.
+   * 
+   * @return an Observable that emits only the very first item emitted by the source Observable, or raises an
+   *         `IllegalArgumentException` if the source Observable is empty
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-first">RxJava Wiki: first()</a>
+   * @see "MSDN: Observable.firstAsync()"
    */
-  def first: Observable[T] = take(1)
+  def first: Observable[T] = {
+    toScalaObservable[T](asJavaObservable.first)
+  }
 
-  /*
-  
-  TODO once https://github.com/Netflix/RxJava/issues/417 is fixed, we can add head and tail methods
+  /**
+   * Returns an Observable that emits only the very first item emitted by the source Observable, or raises an
+   * `IllegalArgumentException` if the source Observable is empty.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/first.png">
+   * 
+   * @return an Observable that emits only the very first item emitted by the source Observable, or raises an
+   *         `IllegalArgumentException` if the source Observable is empty
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-first">RxJava Wiki: first()</a>
+   * @see "MSDN: Observable.firstAsync()"
+   * @see [[Observable.first]]
+   */
+  def head: Observable[T] = first
 
   /**
-   * emits NoSuchElementException("head of empty Observable") if empty
+   * Returns an Observable that emits the last item emitted by the source Observable or notifies observers of
+   * an `IllegalArgumentException` if the source Observable is empty.
+   * 
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/last.png">
+   * 
+   * @return an Observable that emits the last item from the source Observable or notifies observers of an
+   *         error
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observable-Operators#wiki-last">RxJava Wiki: last()</a>
+   * @see "MSDN: Observable.lastAsync()"
    */
-  def head: Observable[T] = {
-    this.take(1).fold[Option[T]](None)((v: Option[T], e: T) => Some(e)).map({
-      case Some(element) => element
-      case None => throw new NoSuchElementException("head of empty Observable")
-    })
+  def last: Observable[T] = {
+    toScalaObservable[T](asJavaObservable.last)
   }
-  
+
   /**
-   * emits an UnsupportedOperationException("tail of empty list") if empty
+   * 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 `IllegalArgumentException`.
+   * 
+   * <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 or no items
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-single-and-singleordefault">RxJava Wiki: single()</a>
+   * @see "MSDN: Observable.singleAsync()"
    */
-  def tail: Observable[T] = ???
-  
-  */
+  def single: Observable[T] = {
+    toScalaObservable[T](asJavaObservable.single)
+  }
 
   /**
    * Returns an Observable that forwards all sequentially distinct items emitted from the source Observable.

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

@@ -53,6 +53,49 @@ class BlockingObservable[+T] private[scala] (val asJava: rx.observables.Blocking
     new WithFilter[T](p, asJava)
   }
 
+  /**
+   * Returns the last item emitted by a specified [[Observable]], or
+   * throws `IllegalArgumentException` if it emits no items.
+   * 
+   * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.last.png">
+   * 
+   * @return the last item emitted by the source [[Observable]]
+   * @throws IllegalArgumentException
+   *             if source contains no elements
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Blocking-Observable-Operators#last-and-lastordefault">RxJava Wiki: last()</a>
+   * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.last.aspx">MSDN: Observable.Last</a>
+   */
+  def last : T = {
+    asJava.last : T
+  }
+
+  /**
+   * Returns the first item emitted by a specified [[Observable]], or
+   * `IllegalArgumentException` if source contains no elements.
+   * 
+   * @return the first item emitted by the source [[Observable]]
+   * @throws IllegalArgumentException
+   *             if source contains no elements
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Blocking-Observable-Operators#first-and-firstordefault">RxJava Wiki: first()</a>
+   * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177.aspx">MSDN: Observable.First</a>
+   */
+  def first : T = {
+    asJava.first : T
+  }
+
+  /**
+   * Returns the first item emitted by a specified [[Observable]], or
+   * `IllegalArgumentException` if source contains no elements.
+   * 
+   * @return the first item emitted by the source [[Observable]]
+   * @throws IllegalArgumentException
+   *             if source contains no elements
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Blocking-Observable-Operators#first-and-firstordefault">RxJava Wiki: first()</a>
+   * @see <a href="http://msdn.microsoft.com/en-us/library/hh229177.aspx">MSDN: Observable.First</a>
+   * @see [[BlockingObservable.first]]
+   */
+  def head : T = first
+
   // last                 -> use toIterable.last
   // lastOrDefault        -> use toIterable.lastOption
   // first                -> use toIterable.head