Merge branch 'master' into cachedthreadscheduler2

Author: jbripley

language-adaptors/rxjava-scala/README.md

@@ -5,14 +5,14 @@ This adaptor allows to use RxJava in Scala with anonymous functions, e.g.
 ```scala
 val o = Observable.interval(200 millis).take(5)
 o.subscribe(n => println("n = " + n))
-Observable(1, 2, 3, 4).reduce(_ + _)
+Observable.items(1, 2, 3, 4).reduce(_ + _)
 ```
 
 For-comprehensions are also supported:
 
 ```scala
-val first = Observable(10, 11, 12)
-val second = Observable(10, 11, 12)
+val first = Observable.items(10, 11, 12)
+val second = Observable.items(10, 11, 12)
 val booleans = for ((n1, n2) <- (first zip second)) yield (n1 == n2)
 ```
 

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)
@@ -620,6 +645,34 @@ class RxScalaDemo extends JUnitSuite {
     o.take(1).subscribe(println(_))
   }
 
+  @Test def createExampleGood2() {
+    import scala.io.{Codec, Source}
+
+    val rxscala = Observable[String](subscriber => {
+      try {
+        val input = new java.net.URL("http://rxscala.github.io/").openStream()
+        subscriber.add(Subscription {
+          input.close()
+        })
+        Source.fromInputStream(input)(Codec.UTF8).getLines()
+          .takeWhile(_ => !subscriber.isUnsubscribed)
+          .foreach(subscriber.onNext(_))
+        if (!subscriber.isUnsubscribed) {
+          subscriber.onCompleted()
+        }
+      }
+      catch {
+        case e: Throwable => if (!subscriber.isUnsubscribed) subscriber.onError(e)
+      }
+    }).subscribeOn(IOScheduler())
+
+    val count = rxscala.flatMap(_.split("\\W+").toSeq.toObservable)
+      .map(_.toLowerCase)
+      .filter(_ == "rxscala")
+      .size
+    println(s"RxScala appears ${count.toBlockingObservable.single} times in http://rxscala.github.io/")
+  }
+
   def output(s: String): Unit = println(s)
 
   /** Subscribes to obs and waits until obs has completed. Note that if you subscribe to

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

@@ -336,6 +336,42 @@ 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.
+   * <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
+   * @param selector a function that combines the pairs of items from the Observable and the Iterable to generate
+   *                 the items to be emitted by the resulting Observable
+   * @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 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))
+  }
+
   /**
    * Returns an Observable formed from this Observable and another Observable by combining
    * corresponding elements using the selector function.
@@ -353,7 +389,7 @@ trait Observable[+T]
    *         their index. Indices start at 0.
    */
   def zipWithIndex: Observable[(T, Int)] = {
-    zip((0 until Int.MaxValue).toObservable)
+    zip(0 until Int.MaxValue)
   }
 
   /**
@@ -3330,10 +3366,19 @@ object Observable {
    * Write the function you pass so that it behaves as an Observable: It should invoke the
    * Subscriber's `onNext`, `onError`, and `onCompleted` methods appropriately.
    *
+   * You can `add` custom [[Subscription]]s to [[Subscriber]]. These [[Subscription]]s will be called
+   * <ul>
+   *   <li>when someone calls `unsubscribe`.</li>
+   *   <li>after `onCompleted` or `onError`.</li>
+   * </ul>
+   *
    * See <a href="http://go.microsoft.com/fwlink/?LinkID=205219">Rx Design Guidelines (PDF)</a> for detailed
    * information.
    *
-   * @tparam T
+   * See `<a href="https://github.com/Netflix/RxJava/blob/master/language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala">RxScalaDemo</a>.createExampleGood`
+   * and `<a href="https://github.com/Netflix/RxJava/blob/master/language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala">RxScalaDemo</a>.createExampleGood2`.
+   *
+   * @param T
    *            the type of the items that this Observable emits
    * @param f
    *            a function that accepts a `Subscriber[T]`, and invokes its `onNext`,

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

@@ -87,7 +87,9 @@ class CompletenessTest extends JUnitSuite {
       "firstOrDefault(T, Func1[_ >: T, Boolean])" -> "[use `.filter(condition).firstOrElse(default)`]",
       "groupBy(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: R])" -> "[use `groupBy` and `map`]",
       "groupByUntil(Func1[_ >: T, _ <: TKey], Func1[_ >: GroupedObservable[TKey, T], _ <: Observable[_ <: TDuration]])" -> "groupByUntil(T => K, (K, Observable[T]) => Observable[Any])",
+      "ignoreElements()" -> "[use `filter(_ => false)`]",
       "lift(Operator[_ <: R, _ >: T])" -> "lift(Subscriber[R] => Subscriber[T])",
+      "limit(Int)" -> "take(Int)",
       "mapWithIndex(Func2[_ >: T, Integer, _ <: R])" -> "[combine `zipWithIndex` with `map` or with a for comprehension]",
       "multicast(Subject[_ >: T, _ <: R])" -> "multicast(Subject[R])",
       "multicast(Func0[_ <: Subject[_ >: T, _ <: TIntermediate]], Func1[_ >: Observable[TIntermediate], _ <: Observable[TResult]])" -> "multicast(() => Subject[R], Observable[R] => Observable[U])",
@@ -118,6 +120,7 @@ class CompletenessTest extends JUnitSuite {
       "skipWhile(Func1[_ >: T, Boolean])" -> "dropWhile(T => Boolean)",
       "skipWhileWithIndex(Func2[_ >: T, Integer, Boolean])" -> unnecessary,
       "skipUntil(Observable[U])" -> "dropUntil(Observable[E])",
+      "single(Func1[_ >: T, Boolean])" -> "[use `filter(predicate).single`]",
       "startWith(T)" -> "[use `item +: o`]",
       "startWith(Array[T])" -> "[use `Observable.items(items) ++ o`]",
       "startWith(Array[T], Scheduler)" -> "[use `Observable.items(items).subscribeOn(scheduler) ++ o`]",
@@ -151,6 +154,8 @@ class CompletenessTest extends JUnitSuite {
       "toSortedList(Func2[_ >: T, _ >: T, Integer])" -> "[Sorting is already done in Scala's collection library, use `.toSeq.map(_.sortWith(f))`]",
       "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])" -> "zipWith(Iterable[U], (T, U) => R)",
 
       // manually added entries for Java static methods
       "average(Observable[Integer])" -> averageProblem,