add methods to BlockingObservable

Author: samuelgruetter

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

@@ -1830,7 +1830,7 @@ class UnitTestSuite extends JUnitSuite {
 
   @Test def testTest() = {
     val a: Observable[Int] = Observable()
-    assertEquals(4, Observable(1, 2, 3, 4).toBlockingObservable.last)
+    assertEquals(4, Observable(1, 2, 3, 4).toBlockingObservable.toIterable.last)
   }
 
 }

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

@@ -23,7 +23,7 @@ import org.junit.{Before, Test, Ignore}
 import org.junit.Assert._
 import rx.lang.scala.concurrency.NewThreadScheduler
 
-//@Ignore // Since this doesn't do automatic testing, don't increase build time unnecessarily
+@Ignore // Since this doesn't do automatic testing, don't increase build time unnecessarily
 class RxScalaDemo extends JUnitSuite {
 
   @Test def intervalExample() {
@@ -229,11 +229,17 @@ class RxScalaDemo extends JUnitSuite {
     waitFor(sharedNumbers)
   }
 
+  @Test def testSingleOption() {
+    assertEquals(None,    Observable(1, 2).toBlockingObservable.singleOption)
+    assertEquals(Some(1), Observable(1)   .toBlockingObservable.singleOption)
+    assertEquals(None,    Observable()    .toBlockingObservable.singleOption)
+  }  
+    
   def output(s: String): Unit = println(s)
 
   // blocks until obs has completed
   def waitFor[T](obs: Observable[T]): Unit = {
-    obs.toBlockingObservable.last
+    obs.toBlockingObservable.toIterable.last
   }
 
 }

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

@@ -22,40 +22,99 @@ class BlockingObservable[+T](val asJava: rx.observables.BlockingObservable[_ <:
   extends AnyVal 
 {
 
+  /**
+   * Invoke a method on each item emitted by the {@link Observable}; block until the Observable
+   * completes.
+   * <p>
+   * NOTE: This will block even if the Observable is asynchronous.
+   * <p>
+   * This is similar to {@link Observable#subscribe(Observer)}, but it blocks. Because it blocks it does
+   * not need the {@link Observer#onCompleted()} or {@link Observer#onError(Throwable)} methods.
+   * <p>
+   * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.forEach.png">
+   *
+   * @param onNext
+   *            the {@link Action1} to invoke for every item emitted by the {@link Observable}
+   * @throws RuntimeException
+   *             if an error occurs
+   */
   def foreach(f: T => Unit): Unit = {
-    asJava.forEach(f)
+    asJava.forEach(f);
   }
 
-  def last: T = {
-    asJava.last() : T // useless ascription because of compiler bug
+  // 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 {@link Iterable} that always returns the item most recently emitted by an {@link Observable}.
+   * <p>
+   * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.mostRecent.png">
+   *
+   * @param initialValue
+   *            the initial value that will be yielded by the {@link Iterable} sequence if the {@link Observable} has not yet emitted an item
+   * @return an {@link Iterable} that on each iteration returns the item that the {@link Observable} has most recently emitted
+   */
+  def mostRecent[U >: T](initialValue: U): Iterable[U] = {
+    val asJavaU = asJava.asInstanceOf[rx.observables.BlockingObservable[U]]
+    asJavaU.mostRecent(initialValue).asScala: Iterable[U] // useless ascription because of compiler bug
   }
 
-  // last(Func1<? super T, Boolean>)
-  // lastOrDefault(T)
-  // lastOrDefault(T, Func1<? super T, Boolean>)
-  // mostRecent(T)
-  // next()
+  /**
+   * Returns an {@link Iterable} that blocks until the {@link Observable} emits another item,
+   * then returns that item.
+   * <p>
+   * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.next.png">
+   *
+   * @return an {@link Iterable} that blocks upon each iteration until the {@link Observable} emits a new item, whereupon the Iterable returns that item
+   */
+  def next: Iterable[T] = {
+    asJava.next().asScala: Iterable[T] // useless ascription because of compiler bug
+  }
 
+  /**
+   * 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">
+   *
+   * @return the single item emitted by the {@link Observable}
+   */
   def single: T = {
-    asJava.single() : T // useless ascription because of compiler bug
+    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}.
+   */
+  def singleOption: Option[T] = {
+    var size: Int = 0
+    var last: Option[T] = None
+    for (t <- toIterable) {
+      size += 1
+      last = Some(t)
+    }
+    if (size == 1) last else None
   }
-  
-  // single(Func1<? super T, Boolean>)
-  
-  // def singleOption: Option[T] = { TODO  }
-  // corresponds to Java's
-  // singleOrDefault(T)
-  
-  // singleOrDefault(BlockingObservable<? extends T>, boolean, T)
-  // singleOrDefault(T, Func1<? super T, Boolean>)
-  // toFuture()
-  
+
+  // TODO toFuture()
+
+  /**
+   * Returns an {@link Iterator} that iterates over all items emitted by this {@link Observable}.
+   */
   def toIterable: Iterable[T] = {
-    asJava.toIterable().asScala : Iterable[T] // useless ascription because of compiler bug
+    asJava.toIterable().asScala: Iterable[T] // useless ascription because of compiler bug
   }
-  
+
+  /**
+   * Returns a {@link List} that contains all items emitted by this {@link Observable}.
+   */
   def toList: List[T] = {
-    asJava.toIterable().asScala.toList : List[T] // useless ascription because of compiler bug
+    asJava.toIterable().asScala.toList: List[T] // useless ascription because of compiler bug
   }
 
-}
+}