add methods and examples

Author: samuelgruetter

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

@@ -73,6 +73,11 @@ object ImplicitFunctionConversions {
     new Func2[A, B, jlang.Boolean] {
       def call(a: A, b: B): jlang.Boolean = f(a, b).booleanValue
     }
+  
+  implicit def scalaFuncNToRxFuncN[R](f: Seq[java.lang.Object] => R): FuncN[R] =
+    new FuncN[R] {
+      def call(args: java.lang.Object*): R = f(args)
+    }
 
   /**
    * Converts a specific function shape (used in takeWhile) to the equivalent Java types with an Rx Func2

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

@@ -199,6 +199,8 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
   def zip[U](that: Observable[U]): Observable[(T, U)] = {
     Observable[(T, U)](JObservable.zip[T, U, (T, U)](this.asJava, that.asJava, (t: T, u: U) => (t, u)))
   }
+    
+  // public static <R> Observable<R> zip(Observable<? extends Observable<?>> ws, final FuncN<? extends R> zipFunction) {
 
   /**
    * Zips this Observable with its indices.
@@ -1068,6 +1070,21 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
     Observable[T](asJava.skip(n))
   }
 
+  /**
+   * Returns an Observable that bypasses all items from the source Observable as long as the specified
+   * condition holds true. Emits all further source items as soon as the condition becomes false.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/skipWhile.png">
+   *
+   * @param predicate
+   *            A function to test each item emitted from the source Observable for a condition.
+   * @return an Observable that emits all items from the source Observable as soon as the condition
+   *         becomes false.
+   */
+  def dropWhile(predicate: T => Boolean): Observable[T] = {
+    Observable[T](asJava.skipWhile(predicate))
+  }
+
   /**
    * Returns an Observable that emits only the first <code>num</code> items emitted by the source
    * Observable.
@@ -1436,6 +1453,56 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
   def product[U >: T](implicit num: Numeric[U]): Observable[U] = {
     fold(num.one)(num.times)
   }
+  
+  /**
+   * TODO doc&test
+   */
+  def firstOrElse[U >: T](default: => U): Observable[U] = {
+    this.materialize.take(1).map((n: Notification[T]) => {
+      if (n.getKind() == rx.Notification.Kind.OnNext)
+        n.getValue
+      else
+        default
+    })
+  }
+  
+  // TODO which of these two find variants do we want?
+  
+  /**
+   * Finds the first element of the list satisfying a predicate, if any.
+   * @param p 
+   *        the predicate used to test elements.
+   * @return an Observable emitting an Option containing the first element in the source 
+   *         Observable that satisfies p, or None if none exists or onError was called.
+   */  
+  def find(p: T => Boolean): Observable[Option[T]] = {
+    this.filter(p).materialize.take(1).map((n: Notification[T]) => {
+      if (n.getKind() == rx.Notification.Kind.OnNext)
+        Some(n.getValue())
+      else
+        None
+    })
+  }
+  
+  /**
+   * Finds the first element of the list satisfying a predicate, if any.
+   * @param p 
+   *        the predicate used to test elements.
+   * @return an Observable emitting an Option containing the first element in the source 
+   *         Observable that satisfies p, or None if none exists.
+   */  
+  private def findWhichTransmitsError(p: T => Boolean): Observable[Option[T]] = {
+    val o: Observable[Notification[Option[T]]] = 
+      this.filter(p).materialize.take(1).map((n: Notification[T]) => {
+      if (n.getKind() == rx.Notification.Kind.OnCompleted)
+        Notification(None)
+      else if (n.getKind() == rx.Notification.Kind.OnNext)
+        Notification(Some(n.getValue()))
+      else 
+        Notification(n.getThrowable())
+    })    
+    o.dematerialize
+  }
 
   /**
    * Converts an Observable into a {@link BlockingObservable} (an Observable with blocking
@@ -1461,6 +1528,7 @@ object Observable {
   import rx.{Observable => JObservable}
   import rx.lang.scala.{Notification, Subscription, Scheduler, Observer}
   import rx.lang.scala.util._
+  import rx.util.functions._
   import rx.lang.scala.ImplicitFunctionConversions._
 
   private[scala] 
@@ -1676,19 +1744,19 @@ object Observable {
   def never: Observable[Nothing] = {
     Observable[Nothing](JObservable.never())
   }
-  
+
   /*
   def apply[T](f: Future[T]): Observable[T] = {
     ??? // TODO convert Scala Future to Java Future
   } 
   */
-  
+
   /*
   def apply[T](f: Future[T], scheduler: Scheduler): Observable[T] = {
     ??? // TODO convert Scala Future to Java Future
   }
   */
-  
+
   /*
   def apply[T](f: Future[T], duration: Duration): Observable[T] = {
     ??? // TODO convert Scala Future to Java Future
@@ -1700,7 +1768,7 @@ object Observable {
    * each time an event is received from one of the source observables, where the aggregation is defined by the given function.
    * <p>
    * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/combineLatest.png">
-   * 
+   *
    * @param o1
    *            The first source observable.
    * @param o2
@@ -1712,14 +1780,57 @@ object Observable {
   // public static <T1, T2, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> combineFunction) 
   // TODO do we want this as an instance method?
   // TODO then decide about combineLatest with > 2 Observables
-  
+
   // TODO what about these two?
   // public static <R> Observable<R> zip(Observable<? extends Observable<?>> ws, final FuncN<? extends R> zipFunction)
   // public static <R> Observable<R> zip(Collection<? extends Observable<?>> ws, FuncN<? extends R> zipFunction)
 
+  /**
+   * Given a Seq of N observables, returns an observable that emits Seqs of N elements each.
+   * The first emitted Seq will contain the first element of each source observable,
+   * the second Seq the second element of each source observable, and so on.
+   * 
+   * @param observables
+   *            A Seq of source Observables
+   * @return an Observable that emits the zipped Seqs
+   */
+  def zip[T](observables: Seq[Observable[T]]): Observable[Seq[T]] = {
+    val f: FuncN[Seq[T]] = (args: Seq[java.lang.Object]) => {
+      val asSeq: Seq[Object] = args.toSeq
+      asSeq.asInstanceOf[Seq[T]]
+    }
+    val list = observables.map(_.asJava).asJava
+    val o = rx.Observable.zip(list, f)
+    Observable[Seq[T]](o)
+  }
+  
+  /**
+   * Given an Observable emitting N source observables, returns an observable that emits Seqs of N elements each.
+   * The first emitted Seq will contain the first element of each source observable,
+   * the second Seq the second element of each source observable, and so on.
+   * 
+   * @param observables
+   *            An Observable emitting N source Observables
+   * @return an Observable that emits the zipped Seqs
+   */
+  def zip[T](observables: Observable[Observable[T]]): Observable[Seq[T]] = {
+    val f: FuncN[Seq[T]] = (args: Seq[java.lang.Object]) => {
+      val asSeq: Seq[Object] = args.toSeq
+      asSeq.asInstanceOf[Seq[T]]
+    }
+    val list = observables.map(_.asJava).asJava
+    val o = rx.Observable.zip(list, f)
+    Observable[Seq[T]](o)
+  }
+  
   def interval(duration: Duration): Observable[Long] = {
     (new Observable[java.lang.Long](JObservable.interval(duration.length, duration.unit))).map(_.longValue())
   }
+  
+  def interval(duration: Duration, scheduler: Scheduler): Observable[Long] = {
+    (new Observable[java.lang.Long](JObservable.interval(duration.length, duration.unit, scheduler))).map(_.longValue())
+  }
+  
 }
 
 // Cannot yet have inner class because of this error message: 
@@ -1776,6 +1887,14 @@ class UnitTestSuite extends JUnitSuite {
     assertEquals(demat.toBlockingObservable.toIterable.toList, List(1, 2, 3))
   }
 
+  @Test def testFind() {
+    assertEquals(Some(3), Observable(1, 3, 5).find(_ >= 2).toBlockingObservable.single)
+    assertEquals(Some(1), Observable(1, 3, 5).find(_ => true).toBlockingObservable.single)
+    assertEquals(None, Observable(1, 3, 5).find(_ > 10).toBlockingObservable.single)
+    assertEquals(None, Observable(new Exception()).find((i: Int) => i > 10).toBlockingObservable.single)
+    assertEquals(None, Observable().find((i: Int) => i > 10).toBlockingObservable.single)
+  }
+  
   @Test def testTest() = {
     val a: Observable[Int] = Observable()
     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() {
@@ -310,6 +310,36 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(1, Observable[Int]().product.toBlockingObservable.single)
   }
 
+  @Test def mapWithIndexExample() {
+    // We don't need mapWithIndex because we already have zipWithIndex, which we can easily
+    // combine with map:
+    Observable("a", "b", "c").zipWithIndex.map(pair => pair._1 + " has index " + pair._2)
+        .toBlockingObservable.foreach(println(_))
+        
+    // Or even nicer with for-comprehension syntax:
+    (for ((letter, index) <- Observable("a", "b", "c").zipWithIndex) yield letter + " has index " + index)
+        .toBlockingObservable.foreach(println(_))
+  }
+  
+  // source Observables are in a List:
+  @Test def zipManySeqExample() {
+    val observables = List(Observable(1, 2), Observable(10, 20), Observable(100, 200))
+    (for (seq <- Observable.zip(observables)) yield seq.mkString("(", ", ", ")"))
+        .toBlockingObservable.foreach(println(_))
+  }
+  
+  // source Observables are in an Observable:
+  @Test def zipManyObservableExample() {
+    val observables = Observable(Observable(1, 2), Observable(10, 20), Observable(100, 200))
+    (for (seq <- Observable.zip(observables)) yield seq.mkString("(", ", ", ")"))
+        .toBlockingObservable.foreach(println(_))
+  }
+  
+  @Test def takeFirstWithCondition() {
+    val condition: Int => Boolean = _ >= 3
+    assertEquals(3, Observable(1, 2, 3, 4).dropWhile(!condition(_)).take(1).toBlockingObservable.single)
+  }
+  
   def output(s: String): Unit = println(s)
 
   // blocks until obs has completed

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

@@ -16,6 +16,9 @@ class CompletenessTest extends JUnitSuite {
   val averageProblem = "[We can't have a general average method because Scala's Numeric does not have " +
      "scalar multiplication (we would need to calculate (1.0/numberOfElements)*sum). " + 
      "You can use fold instead to accumulate sum and numberOfElements and divide at the end.]"
+     
+  val commentForFirst = "[use take(1)]"
+  val commentForFirstWithPredicate = "[use .dropWhile(!condition).take(1) or use find(condition)]"
 
   val correspondence = defaultMethodCorrespondence ++ Map(
       // manually added entries for Java instance methods
@@ -25,8 +28,13 @@ class CompletenessTest extends JUnitSuite {
       "buffer(Long, Long, TimeUnit)" -> "buffer(Duration, Duration)",
       "buffer(Long, Long, TimeUnit, Scheduler)" -> "buffer(Duration, Duration, Scheduler)",
       "dematerialize()" -> "dematerialize(<:<[T, Notification[U]])",
+      "first()" -> commentForFirst,
+      "first(Func1[_ >: T, Boolean])" -> commentForFirstWithPredicate,
+      //"firstOrDefault(T)" -> 
+      "firstOrDefault(Func1[_ >: T, Boolean], T)" -> "[use find and map instead]", // TODO maybe firstOrElse method?
       "groupBy(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: R])" -> "groupBy(T => K)",
       "mapMany(Func1[_ >: T, _ <: Observable[_ <: R]])" -> "flatMap(T => Observable[R])",
+      "mapWithIndex(Func2[_ >: T, Integer, _ <: R])" -> "[combine zipWithIndex with map or with a for comprehension]",
       "onErrorResumeNext(Func1[Throwable, _ <: Observable[_ <: T]])" -> "onErrorResumeNext(Throwable => Observable[U])",
       "onErrorResumeNext(Observable[_ <: T])" -> "onErrorResumeNext(Observable[U])",
       "onErrorReturn(Func1[Throwable, _ <: T])" -> "onErrorReturn(Throwable => U)",
@@ -39,6 +47,8 @@ class CompletenessTest extends JUnitSuite {
       "skipWhile(Func1[_ >: T, Boolean])" -> "dropWhile(T => Boolean)",
       "skipWhileWithIndex(Func2[_ >: T, Integer, Boolean])" -> unnecessary,
       "startWith(Iterable[T])" -> "[unnecessary because we can just use ++ instead]",
+      "takeFirst()" -> commentForFirst,
+      "takeFirst(Func1[_ >: T, Boolean])" -> commentForFirstWithPredicate,
       "takeLast(Int)" -> "takeRight(Int)",
       "toList()" -> "toSeq",
       "toSortedList()" -> unnecessary,
@@ -70,7 +80,9 @@ class CompletenessTest extends JUnitSuite {
       "sumLongs(Observable[Long])" -> "sum(Numeric[U])",
       "synchronize(Observable[T])" -> "synchronize",
       "switchDo(Observable[_ <: Observable[_ <: T]])" -> "switch",
-      "zip(Observable[_ <: T1], Observable[_ <: T2], Func2[_ >: T1, _ >: T2, _ <: R])" -> "[use instance method zip and map]"
+      "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]"
   ) ++ List.iterate("T", 9)(s => s + ", T").map(
       // all 9 overloads of startWith:
       "startWith(" + _ + ")" -> "[unnecessary because we can just use ++ instead]"