Merge pull request #1609 from zsxwing/rxscala-backpressure

RxScala: Add backpressure support

Author: benjchristensen

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

@@ -814,6 +814,48 @@ class RxScalaDemo extends JUnitSuite {
     println(s"RxScala appears ${count.toBlocking.single} times in http://rxscala.github.io/")
   }
 
+  @Test def createExampleWithBackpressure() {
+    val o = Observable {
+      subscriber: Subscriber[String] => {
+        var emitted = 0
+        subscriber.setProducer(n => {
+            val intN = if (n >= 10) 10 else n.toInt
+            (0 until intN)
+              .takeWhile(_ => emitted < 10 && !subscriber.isUnsubscribed)
+              .foreach {
+              i =>
+                emitted += 1
+                subscriber.onNext(s"item ${emitted}")
+            }
+            if (emitted == 10 && !subscriber.isUnsubscribed) {
+              subscriber.onCompleted()
+            }
+        })
+      }
+    }.subscribeOn(IOScheduler()) // Use `subscribeOn` to make sure `Producer` will run in the same Scheduler
+    o.observeOn(ComputationScheduler()).subscribe(new Subscriber[String] {
+      override def onStart() {
+        println("Request a new one at the beginning")
+        request(1)
+      }
+
+      override def onNext(v: String) {
+        println("Received " + v)
+        println("Request a new one after receiving " + v)
+        request(1)
+      }
+
+      override def onError(e: Throwable) {
+        e.printStackTrace()
+      }
+
+      override def onCompleted() {
+        println("Done")
+      }
+    })
+    Thread.sleep(10000)
+  }
+
   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/Producer.scala

@@ -0,0 +1,55 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.lang.scala
+
+import rx.{Producer => JProducer}
+
+
+trait Producer {
+
+  // Java calls XXX, Scala receives XXX
+  private[scala] val asJavaProducer: JProducer = new JProducer {
+    def request(n: Long): Unit = {
+      Producer.this.request(n)
+    }
+  }
+
+  def request(n: Long): Unit = {}
+}
+
+object Producer {
+  /**
+   * Scala calls XXX; Java receives XXX.
+   */
+  private[scala] def apply[T](producer: JProducer): Producer = {
+    new Producer {
+      override val asJavaProducer = producer
+
+      override def request(n: Long): Unit = {
+        producer.request(n)
+      }
+    }
+  }
+
+  def apply[T](producer: Long => Unit): Producer = {
+    // Java calls XXX; Scala receives XXX.
+    Producer(new JProducer {
+      override def request(n: Long): Unit = {
+        producer(n)
+      }
+    })
+  }
+}

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

@@ -4,10 +4,12 @@ trait Subscriber[-T] extends Observer[T] with Subscription {
 
   self =>
 
-  private [scala] val asJavaSubscriber: rx.Subscriber[_ >: T] = new rx.Subscriber[T] {
-    def onNext(value: T): Unit = self.onNext(value)
-    def onError(error: Throwable): Unit = self.onError(error)
-    def onCompleted(): Unit = self.onCompleted()
+  private [scala] val asJavaSubscriber: rx.Subscriber[_ >: T] = new rx.Subscriber[T] with SubscriberAdapter[T] {
+    override def onStart(): Unit = self.onStart()
+    override def onNext(value: T): Unit = self.onNext(value)
+    override def onError(error: Throwable): Unit = self.onError(error)
+    override def onCompleted(): Unit = self.onCompleted()
+    override def requestMore(n: Long): Unit = request(n)
   }
 
   private [scala] override val asJavaObserver: rx.Observer[_ >: T] = asJavaSubscriber
@@ -37,6 +39,24 @@ trait Subscriber[-T] extends Observer[T] with Subscription {
     asJavaSubscriber.isUnsubscribed()
   }
 
+  def onStart(): Unit = {
+    asJavaSubscriber.onStart()
+  }
+
+  protected final def request(n: Long): Unit = {
+    asJavaSubscriber match {
+      case s: SubscriberAdapter[T] => s.requestMore(n)
+      case _ => throw new rx.exceptions.MissingBackpressureException()
+    }
+  }
+
+  def setProducer(producer: Producer): Unit = {
+    asJavaSubscriber.setProducer(producer.asJavaProducer)
+  }
+
+  def setProducer(producer: Long => Unit): Unit = {
+    asJavaSubscriber.setProducer(Producer(producer).asJavaProducer)
+  }
 }
 
 object Subscriber extends ObserverFactoryMethods[Subscriber] {
@@ -71,3 +91,8 @@ object Subscriber extends ObserverFactoryMethods[Subscriber] {
     })
   }
 }
+
+sealed trait SubscriberAdapter[T] extends rx.Subscriber[T] {
+  // Add a method to expose the protected `request` method
+  def requestMore(n: Long): Unit
+}