add zipWithIndex, sum, product, and an explanation why no average

Author: samuelgruetter

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

@@ -198,8 +198,20 @@ 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)))
-  }  
+  }
 
+  /**
+   * Zips this Observable with its indices.
+   * 
+   * @return An Observable emitting pairs consisting of all elements of this Observable paired with 
+   *         their index. Indices start at 0.
+   */
+  def zipWithIndex: Observable[(T, Int)] = {
+    val fScala: (T, Integer) => (T, Int) = (elem: T, index: Integer) => (elem, index)
+    val fJava : Func2[_ >: T, Integer, _ <: (T, Int)] = fScala
+    Observable[(T, Int)](asJava.mapWithIndex[(T, Int)](fJava))
+  }
+    
   /**
    * Creates an Observable which produces buffers of collected values.
    * 
@@ -1384,7 +1396,27 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
   def throttleLast(intervalDuration: Duration, scheduler: Scheduler): Observable[T] = {
     Observable[T](asJava.throttleLast(intervalDuration.length, intervalDuration.unit, scheduler))
   }
+
+  /**
+   * Returns an Observable that sums up the elements of this Observable.
+   * 
+   * @return an Observable emitting the sum of all the elements of the source Observable
+   *         as its single item.
+   */
+  def sum[U >: T](implicit num: Numeric[U]): Observable[U] = {
+    fold(num.zero)(num.plus)
+  }
 
+  /**
+   * Returns an Observable that multiplies up the elements of this Observable.
+   * 
+   * @return an Observable emitting the product of all the elements of the source Observable
+   *         as its single item.
+   */
+  def product[U >: T](implicit num: Numeric[U]): Observable[U] = {
+    fold(num.one)(num.times)
+  }
+
   /**
    * Converts an Observable into a {@link BlockingObservable} (an Observable with blocking
    * operators).

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

@@ -233,7 +233,33 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(None,    Observable(1, 2).toBlockingObservable.singleOption)
     assertEquals(Some(1), Observable(1)   .toBlockingObservable.singleOption)
     assertEquals(None,    Observable()    .toBlockingObservable.singleOption)
-  }  
+  }
+  
+  // We can't put a general average method into Observable.scala, because Scala's Numeric
+  // does not have scalar multiplication (we would need to calculate (1.0/numberOfElements)*sum)
+  def doubleAverage(o: Observable[Double]): Observable[Double] = {
+    for ((finalSum, finalCount) <- o.fold((0.0, 0))({case ((sum, count), elem) => (sum+elem, count+1)}))
+      yield finalSum / finalCount
+  }
+  
+  @Test def averageExample() {
+    println(doubleAverage(Observable()).toBlockingObservable.single)
+    println(doubleAverage(Observable(0)).toBlockingObservable.single)
+    println(doubleAverage(Observable(4.44)).toBlockingObservable.single)
+    println(doubleAverage(Observable(1, 2, 3.5)).toBlockingObservable.single)
+  }
+  
+  @Test def testSum() {
+    assertEquals(10, Observable(1, 2, 3, 4).sum.toBlockingObservable.single)
+    assertEquals(6, Observable(4, 2).sum.toBlockingObservable.single)
+    assertEquals(0, Observable[Int]().sum.toBlockingObservable.single)
+  }
+  
+  @Test def testProduct() {
+    assertEquals(24, Observable(1, 2, 3, 4).product.toBlockingObservable.single)
+    assertEquals(8, Observable(4, 2).product.toBlockingObservable.single)
+    assertEquals(1, Observable[Int]().product.toBlockingObservable.single)
+  }
 
   def output(s: String): Unit = println(s)
 

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

@@ -12,6 +12,10 @@ import java.lang.reflect.Modifier
 class CompletenessTest extends JUnitSuite {
 
   val unnecessary = "[considered unnecessary in Scala land]"
+    
+  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 correspondence = defaultMethodCorrespondence ++ Map(
       // manually added entries for Java instance methods
@@ -44,6 +48,10 @@ class CompletenessTest extends JUnitSuite {
       "window(Long, Long, TimeUnit, Scheduler)" -> "window(Duration, Duration, Scheduler)",
 
       // manually added entries for Java static methods
+      "average(Observable[Integer])" -> averageProblem,
+      "averageDoubles(Observable[Double])" -> averageProblem,
+      "averageFloats(Observable[Float])" -> averageProblem,
+      "averageLongs(Observable[Long])" -> averageProblem,
       "create(OnSubscribeFunc[T])" -> "apply(Observer[T] => Subscription)",
       "defer(Func0[_ <: Observable[_ <: T]])" -> "defer(=> Observable[T])",
       "empty()" -> "apply(T*)",
@@ -55,6 +63,10 @@ class CompletenessTest extends JUnitSuite {
       "range(Int, Int)" -> "apply(Range)",
       "sequenceEqual(Observable[_ <: T], Observable[_ <: T])" -> "[use (first zip second) map (p => p._1 == p._2)]",
       "sequenceEqual(Observable[_ <: T], Observable[_ <: T], Func2[_ >: T, _ >: T, Boolean])" -> "[use (first zip second) map (p => equality(p._1, p._2))]",
+      "sum(Observable[Integer])" -> "sum(Numeric[U])",
+      "sumDoubles(Observable[Double])" -> "sum(Numeric[U])",
+      "sumFloats(Observable[Float])" -> "sum(Numeric[U])",
+      "sumLongs(Observable[Long])" -> "sum(Numeric[U])",
       "switchDo(Observable[_ <: Observable[_ <: T]])" -> "switch",
       "synchronize(Observable[_ <: T])" -> "synchronize",
       "zip(Observable[_ <: T1], Observable[_ <: T2], Func2[_ >: T1, _ >: T2, _ <: R])" -> "[use instance method zip and map]"