Merge pull request #197 from zsxwing/rxjava-1.1.6

Bump to RxJava 1.1.6

build.sbt

@@ -19,7 +19,7 @@ crossScalaVersions in ThisBuild := Seq("2.10.6", "2.11.8", "2.12.0-M4")
 parallelExecution in Test := false
 
 libraryDependencies ++= Seq(
-  "io.reactivex" % "rxjava" % "1.1.5",
+  "io.reactivex" % "rxjava" % "1.1.6",
   "org.mockito" % "mockito-core" % "1.9.5" % "test",
   "junit" % "junit" % "4.11" % "test",
   "org.scalatest" %% "scalatest" % "2.2.6" % "test")

examples/src/test/scala/examples/RxScalaDemo.scala

@@ -1896,4 +1896,23 @@ class RxScalaDemo extends JUnitSuite {
       println(s"onTerminateDetach: isUnsubscribed=${s.isUnsubscribed}, weakReference=${weakReference.get}")
     }
   }
+
+  @Test def rebatchRequestsExample(): Unit = {
+    val o = (1 to 100).toObservable
+    o.doOnRequest(r => println(s"Requesting $r via rebatchRequests"))
+      .rebatchRequests(10)
+      .doOnRequest(r => println(s"Requesting $r via Subscriber"))
+      .subscribe(new Subscriber[Int]() {
+        override def onStart(): Unit = request(1)
+
+        override def onNext(value: Int): Unit = {
+          println(s"Receive $value")
+          request(1)
+        }
+
+        override def onError(error: Throwable): Unit = error.printStackTrace()
+
+        override def onCompleted(): Unit = println("Done")
+      })
+  }
 }

src/main/scala/rx/lang/scala/Observable.scala

@@ -1361,6 +1361,26 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.unsubscribeOn(scheduler))
   }
 
+  /**
+   * $experimental Returns an [[Observable]] that requests `n` initially from the upstream and then 75% of `n` subsequently after 75% of `n` values have
+   * been emitted to the downstream.
+   *
+   * This operator allows preventing the downstream to trigger unbounded mode via `request(Long.MaxValue)` or compensate for the per-item
+   * overhead of small and frequent requests.
+   *
+   * ===Backpressure:===
+   * The operator expects backpressure from upstream and honors backpressure from downstream.</dd>
+   *
+   * $noDefaultScheduler
+   *
+   * @param n the initial request amount, further request will happen after 75% of this value
+   * @return the [[Observable]] that rebatches request amounts from downstream
+   */
+  @Experimental
+  def rebatchRequests(n: Int): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.rebatchRequests(n))
+  }
+
   /**
    * Asynchronously notify [[rx.lang.scala.Observer]]s on the specified [[rx.lang.scala.Scheduler]].
    *