Rename to 'zipWith'; add 'zip(iterable)'; add examples

Author: zsxwing

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

@@ -448,6 +448,31 @@ class RxScalaDemo extends JUnitSuite {
       .toBlockingObservable.foreach(println(_))
   }
 
+  /**
+   * This is a bad way of using `zip` with an `Iterable`: even if the consumer unsubscribes,
+   * some elements may still be pulled from `Iterable`.
+   */
+  @Test def zipWithIterableBadExample() {
+    val o1 = Observable.interval(100 millis, IOScheduler()).map(_ * 100).take(3)
+    val o2 = Observable.from(0 until Int.MaxValue).doOnEach(i => println(i + " from o2"))
+    o1.zip(o2).toBlockingObservable.foreach(println(_))
+  }
+
+  /**
+   * This is a good way of using `zip` with an `Iterable`: if the consumer unsubscribes,
+   * no more elements will be pulled from `Iterable`.
+   */
+  @Test def zipWithIterableGoodExample() {
+    val o1 = Observable.interval(100 millis, IOScheduler()).map(_ * 100).take(3)
+    val iter = (0 until Int.MaxValue).view.map {
+      i => {
+        println(i + " from iter")
+        i
+      }
+    }
+    o1.zip(iter).toBlockingObservable.foreach(println(_))
+  }
+
   @Test def takeFirstWithCondition() {
     val condition: Int => Boolean = _ >= 3
     assertEquals(3, List(1, 2, 3, 4).toObservable.filter(condition).first.toBlockingObservable.single)

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

@@ -336,6 +336,22 @@ trait Observable[+T]
     zipWith(that, (t: T, u: U) => (t, u))
   }
 
+  /**
+   * Returns an Observable formed from `this` Observable and `other` Iterable by combining
+   * corresponding elements in pairs.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.i.png">
+   * <p>
+   * Note that the `other` Iterable is evaluated as items are observed from the source Observable; it is
+   * not pre-consumed. This allows you to zip infinite streams on either side.
+   *
+   * @param other the Iterable sequence
+   * @return an Observable that pairs up values from the source Observable and the `other` Iterable.
+   */
+  def zip[U](other: Iterable[U]): Observable[(T, U)] = {
+    zipWith(other, (t: T, u: U) => (t, u))
+  }
+
   /**
    * Returns an Observable that emits items that are the result of applying a specified function to pairs of
    * values, one each from the source Observable and a specified Iterable sequence.
@@ -351,7 +367,7 @@ trait Observable[+T]
    * @return an Observable that pairs up values from the source Observable and the `other` Iterable
    *         sequence and emits the results of `selector` applied to these pairs
    */
-  def zip[U, R](other: Iterable[U], selector: (T, U) => R): Observable[R] = {
+  def zipWith[U, R](other: Iterable[U], selector: (T, U) => R): Observable[R] = {
     val thisJava = asJavaObservable.asInstanceOf[rx.Observable[T]]
     toScalaObservable[R](thisJava.zip(other.asJava, selector))
   }
@@ -373,7 +389,7 @@ trait Observable[+T]
    *         their index. Indices start at 0.
    */
   def zipWithIndex: Observable[(T, Int)] = {
-    zip((0 until Int.MaxValue), (t: T, index: Int) => (t, index))
+    zip(0 until Int.MaxValue)
   }
 
   /**

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

@@ -155,7 +155,7 @@ class CompletenessTest extends JUnitSuite {
       "window(Long, Long, TimeUnit)" -> "window(Duration, Duration)",
       "window(Long, Long, TimeUnit, Scheduler)" -> "window(Duration, Duration, Scheduler)",
       "zip(Observable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zipWith(Observable[U], (T, U) => R)",
-      "zip(Iterable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zip(Iterable[U], (T, U) => R)",
+      "zip(Iterable[_ <: T2], Func2[_ >: T, _ >: T2, _ <: R])" -> "zipWith(Iterable[U], (T, U) => R)",
 
       // manually added entries for Java static methods
       "average(Observable[Integer])" -> averageProblem,