Implement SyncOnSubscribe and AsyncOnSubscribe

src/main/scala/rx/lang/scala/Notification.scala

@@ -63,6 +63,16 @@ sealed trait Notification[+T] {
   }
 
   def apply(observer: Observer[T]): Unit = accept(observer)
+
+  // TODO: Should this be public or private to rx.lang.scala?
+  def map[U](f: T => U): Notification[U] = {
+    this match {
+      case Notification.OnNext(value) => Notification.OnNext(f(value))
+      // TODO: Or do we want to cast here? Notification.OnError[T] is not a Notification[U]
+      case Notification.OnError(error) => Notification.OnError(error)
+      case Notification.OnCompleted => Notification.OnCompleted
+    }
+  }
 }
 
 /**

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

@@ -19,7 +19,8 @@ package rx.lang.scala
 import rx.annotations.{Beta, Experimental}
 import rx.exceptions.OnErrorNotImplementedException
 import rx.functions.FuncN
-import rx.lang.scala.observables.{ConnectableObservable, ErrorDelayingObservable}
+import rx.lang.scala.observables.{AsyncOnSubscribe, ConnectableObservable, ErrorDelayingObservable, SyncOnSubscribe}
+
 import scala.concurrent.duration
 import java.util
 
@@ -4919,6 +4920,52 @@ object Observable {
   }
   */
 
+  /**
+    * Returns an Observable that respects the back-pressure semantics. When the returned Observable is
+    * subscribed to it will initiate the given [[observables.SyncOnSubscribe]]'s life cycle for
+    * generating events.
+    *
+    * Note: the `SyncOnSubscribe` provides a generic way to fulfill data by iterating
+    * over a (potentially stateful) function (e.g. reading data off of a channel, a parser). If your
+    * data comes directly from an asynchronous/potentially concurrent source then consider using [[observables.AsyncOnSubscribe]].
+    *
+    * $supportBackpressure
+    *
+    * $noDefaultScheduler
+    *
+    * @tparam T the type of the items that this Observable emits
+    * @tparam S the state type
+    * @param syncOnSubscribe
+    *          an implementation of `SyncOnSubscribe`. There are many creation methods on the object for convenience.
+    * @return an Observable that, when a [[Subscriber]] subscribes to it, will use the specified `SyncOnSubscribe` to generate events
+    * @see [[observables.SyncOnSubscribe.stateful]]
+    * @see [[observables.SyncOnSubscribe.singleState]]
+    * @see [[observables.SyncOnSubscribe.stateless]]
+    */
+  @Experimental
+  def create[S,T](syncOnSubscribe: SyncOnSubscribe[S,T]): Observable[T] = toScalaObservable[T](rx.Observable.create(syncOnSubscribe))
+
+  /**
+    * Returns an Observable that respects the back-pressure semantics. When the returned Observable is
+    * subscribed to it will initiate the given [[observables.AsyncOnSubscribe]]'s life cycle for
+    * generating events.
+    *
+    * $supportBackpressure
+    *
+    * $noDefaultScheduler
+    *
+    * @tparam T the type of the items that this Observable emits
+    * @tparam S the state type
+    * @param asyncOnSubscribe
+    *          an implementation of `AsyncOnSubscribe`. There are many creation methods on the object for convenience.
+    * @return an Observable that, when a [[Subscriber]] subscribes to it, will use the specified `AsyncOnSubscribe` to generate events
+    * @see [[observables.AsyncOnSubscribe.stateful]]
+    * @see [[observables.AsyncOnSubscribe.singleState]]
+    * @see [[observables.AsyncOnSubscribe.stateless]]
+    */
+  @Experimental
+  def create[S,T](asyncOnSubscribe: AsyncOnSubscribe[S,T]): Observable[T] = toScalaObservable[T](rx.Observable.create(asyncOnSubscribe))
+
   /**
    * Returns an Observable that will execute the specified function when someone subscribes to it.
    *

src/main/scala/rx/lang/scala/observables/AsyncOnSubscribe.scala

@@ -0,0 +1,78 @@
+package rx.lang.scala.observables
+
+import rx.annotations.Experimental
+import rx.lang.scala.{Notification, Observable}
+
+/**
+  * An utility class to create `Observable`s that start acting when subscribed to and responds
+  * correctly to back pressure requests from subscribers.
+  *
+  * Semantics:
+  * * `generator` is called to provide an initial state on each new subscription
+  * * `next` is called with the last state and a `requested` amount of items to provide a new state
+  *     and an `Observable` that (potentially asynchronously) emits up to `requested` items.
+  * * `onUnsubscribe` is called with the state provides by the last next when the observer unsubscribes
+  */
+object AsyncOnSubscribe {
+
+  /**
+    * Alias for [[AsyncOnSubscribe.stateful]]
+    * @see [[AsyncOnSubscribe.stateful]]
+    */
+  @Experimental
+  def apply[S,T](generator: () => S)(next: (S, Long) => (Notification[Observable[T]], S), onUnsubscribe: S => Unit = (_:S) => ()): AsyncOnSubscribe[S,T] =
+    stateful[S, T](generator)(next, onUnsubscribe)
+
+  /**
+    * Generates a stateful [[AsyncOnSubscribe]]
+    *
+    * @tparam T the type of the generated values
+    * @tparam S the type of the associated state with each Subscriber
+    * @param generator generates the initial state value
+    * @param next produces observables which contain data for the stream
+    * @param onUnsubscribe clean up behavior
+    */
+  @Experimental
+  def stateful[S, T](generator: () => S)(next: (S, Long) => (Notification[Observable[T]], S), onUnsubscribe: S => Unit = (_:S) => ()): AsyncOnSubscribe[S,T] = {
+    // The anonymous class shadows these names
+    val nextF = next
+    val onUnsubscribeF = onUnsubscribe
+
+    new rx.observables.AsyncOnSubscribe[S,T] {
+      import rx.lang.scala.JavaConversions._
+      override def generateState(): S = generator()
+      override def next(state: S, requested: Long, observer: rx.Observer[rx.Observable[_ <: T]]): S =
+        nextF(state, requested) match {
+          case (notification, nextState) =>
+            toJavaNotification(notification.map(toJavaObservable)).accept(observer)
+            nextState
+        }
+      override def onUnsubscribe(state: S): Unit = onUnsubscribeF(state)
+    }
+  }
+
+  /**
+    * Generates a [[AsyncOnSubscribe]] which does not generate a new state in `next`
+    *
+    * @tparam T the type of the generated values
+    * @tparam S the type of the associated state with each Subscriber
+    * @param generator generates the state value
+    * @param next produces observables which contain data for the stream
+    * @param onUnsubscribe clean up behavior
+    */
+  @Experimental
+  def singleState[S, T](generator: () => S)(next: (S, Long) => Notification[Observable[T]], onUnsubscribe: S => Unit = (_:S) => ()): AsyncOnSubscribe[S,T] =
+    stateful[S, T](generator)((s,r) => (next(s,r), s), onUnsubscribe)
+
+  /**
+    * Generates a stateless [[AsyncOnSubscribe]], useful when the state is closed over in `next` or the `SyncOnSubscribe` inherently does not have a state
+    *
+    * @tparam T the type of the generated values
+    * @param next produces observables which contain data for the stream
+    * @param onUnsubscribe clean up behavior
+    */
+  @Experimental
+  def stateless[T](next: Long => Notification[Observable[T]], onUnsubscribe: () => Unit = () => ()) =
+    stateful[Unit, T](() => ())((_,r) => (next(r), ()), _ => onUnsubscribe())
+
+}

src/main/scala/rx/lang/scala/observables/SyncOnSubscribe.scala

@@ -0,0 +1,77 @@
+package rx.lang.scala.observables
+
+import rx.annotations.Experimental
+import rx.lang.scala.{Notification, Observable}
+
+/**
+  * An utility class to create `Observable`s that start acting when subscribed to and responds
+  * correctly to back pressure requests from subscribers.
+  *
+  * Semantics:
+  * * `generator` is called to provide an initial state on each new subscription
+  * * `next` is called with the last state to provide a data item and a new state for the next `next` call
+  * * `onUnsubscribe` is called with the state provides by the last next when the observer unsubscribes
+  */
+object SyncOnSubscribe {
+
+  /**
+    * Alias for [[SyncOnSubscribe.stateful]]
+    * @see [[SyncOnSubscribe.stateful]]
+    */
+  @Experimental
+  def apply[S, T](generator: () => S)(next: S => (Notification[T], S), onUnsubscribe: S => Unit = (_:S) => ()): SyncOnSubscribe[S,T] =
+    stateful[S, T](generator)(next, onUnsubscribe)
+
+  /**
+    * Generates a stateful [[SyncOnSubscribe]]
+    *
+    * @tparam T the type of the generated values
+    * @tparam S the type of the associated state with each Subscriber
+    * @param generator generates the initial state value
+    * @param next produces data for the stream
+    * @param onUnsubscribe clean up behavior
+    */
+  @Experimental
+  def stateful[S, T](generator: () => S)(next: S => (Notification[T], S), onUnsubscribe: S => Unit = (_:S) => ()): SyncOnSubscribe[S,T] = {
+    // The anonymous class shadows these names
+    val nextF = next
+    val onUnsubscribeF = onUnsubscribe
+
+    new rx.observables.SyncOnSubscribe[S,T] {
+      import rx.lang.scala.JavaConversions._
+      override def generateState(): S = generator()
+      override def next(state: S, observer: rx.Observer[_ >: T]): S =
+        nextF(state) match {
+          case (notification, nextState) =>
+            toJavaNotification(notification).accept(observer)
+            nextState
+        }
+      override def onUnsubscribe(state: S): Unit = onUnsubscribeF(state)
+    }
+  }
+
+  /**
+    * Generates a [[SyncOnSubscribe]] which does not generate a new state in `next`
+    *
+    * @tparam T the type of the generated values
+    * @tparam S the type of the associated state with each Subscriber
+    * @param generator generates the state value
+    * @param next produces data for the stream
+    * @param onUnsubscribe clean up behavior
+    */
+  @Experimental
+  def singleState[S,T](generator: () => S)(next: S => Notification[T], onUnsubscribe: S => Unit = (_:S) => ()): SyncOnSubscribe[S,T] =
+    apply[S, T](generator)(s => (next(s),s), onUnsubscribe)
+
+  /**
+    * Generates a stateless [[SyncOnSubscribe]], useful when the state is closed over in `next` or the `SyncOnSubscribe` inherently does not have a state
+    *
+    * @tparam T the type of the generated values
+    * @param next produces data for the stream
+    * @param onUnsubscribe clean up behavior
+    */
+  @Experimental
+  def stateless[T](next: () => Notification[T], onUnsubscribe: () => Unit = () => ()) =
+    apply[Unit, T](() => ())(_ => (next(), ()), _ => onUnsubscribe())
+
+}

src/main/scala/rx/lang/scala/observables/package.scala

@@ -24,4 +24,7 @@ package rx.lang.scala
  * in Scala, because we use a pair `(key, observable)` instead of `GroupedObservable`
  * and a pair `(startFunction, observable)` instead of `ConnectableObservable`. 
  */
-package object observables {}
+package object observables {
+  type SyncOnSubscribe[S, +T] = rx.observables.SyncOnSubscribe[S, _ <: T]
+  type AsyncOnSubscribe[S, +T] = rx.observables.AsyncOnSubscribe[S, _ <: T]
+}

src/test/scala-2.11/rx/lang/scala/completeness/ObservableCompletenessKit.scala

@@ -198,8 +198,8 @@ class ObservableCompletenessKit extends CompletenessKit {
     // manually added entries for Java static methods
     "amb(Iterable[_ <: Observable[_ <: T]])" -> "amb(Observable[T]*)",
     "create(OnSubscribe[T])" -> "apply(Subscriber[T] => Unit)",
-    "create(SyncOnSubscribe[S, T])" -> "[TODO]",
-    "create(AsyncOnSubscribe[S, T])" -> "[TODO]",
+    "create(SyncOnSubscribe[S, T])" -> "create(SyncOnSubscribe[S, T])",
+    "create(AsyncOnSubscribe[S, T])" -> "create(AsyncOnSubscribe[S, T])",
     "combineLatest(Observable[_ <: T1], Observable[_ <: T2], Func2[_ >: T1, _ >: T2, _ <: R])" -> "combineLatestWith(Observable[U])((T, U) => R)",
     "combineLatest(List[_ <: Observable[_ <: T]], FuncN[_ <: R])" -> "combineLatest(Iterable[Observable[T]])(Seq[T] => R)",
     "combineLatest(Iterable[_ <: Observable[_ <: T]], FuncN[_ <: R])" -> "combineLatest(Iterable[Observable[T]])(Seq[T] => R)",

src/test/scala/rx/lang/scala/observables/AsyncOnSubscribeTests.scala

@@ -0,0 +1,65 @@
+package rx.lang.scala.observables
+
+import org.junit.Test
+import org.junit.Assert._
+import org.scalatest.junit.JUnitSuite
+import rx.lang.scala.observers.TestSubscriber
+import rx.lang.scala.{Notification, Observable}
+
+class AsyncOnSubscribeTests extends JUnitSuite {
+
+  @Test
+  def testStateful(): Unit = {
+    val last = 2000L
+    val sut = Observable.create(AsyncOnSubscribe(() => 0L)((count,demand) =>
+      if(count > last)
+        (Notification.OnCompleted, count)
+      else {
+        val max = math.max(count + demand, last)
+        val next = Observable.from(count to max)
+        (Notification.OnNext(next), max+1)
+      }
+    ))
+    assertEquals((0L to last).toList, sut.toBlocking.toList)
+  }
+
+  @Test
+  def testStateless(): Unit = {
+    val sut = Observable.create(AsyncOnSubscribe.stateless(r => Notification.OnNext(Observable.just(42).repeat(r))))
+    assertEquals(List(42,42,42,42), sut.take(4).toBlocking.toList)
+  }
+
+  @Test
+  def testSingleState(): Unit = {
+    val random = math.random
+    val sut = Observable.create(AsyncOnSubscribe.singleState(() => random)((s,r) => Notification.OnNext(Observable.just(random.toString).repeat(r))))
+    assertEquals(List(random.toString, random.toString), sut.take(2).toBlocking.toList)
+  }
+
+  @Test
+  def testUnsubscribe(): Unit = {
+    val sideEffect = new java.util.concurrent.atomic.AtomicBoolean(false)
+    val sut = Observable.create(AsyncOnSubscribe(() => ())((s,r) => (Notification.OnCompleted, s), onUnsubscribe = s => sideEffect.set(true)))
+    sut.foreach(_ => ())
+    assertEquals(true, sideEffect.get())
+  }
+
+  @Test
+  def testError(): Unit = {
+    val e = new IllegalStateException("Oh noes")
+    val sut = Observable.create(AsyncOnSubscribe(() => 0)((s,_) => (if(s>2) Notification.OnNext(Observable.just(s)) else Notification.OnError(e), s+1)))
+    val testSubscriber = TestSubscriber[Int]()
+    sut.subscribe(testSubscriber)
+    testSubscriber.assertError(e)
+  }
+
+  @Test
+  // Ensure that the generator is executed for each subscription
+  def testGenerator(): Unit = {
+    val sideEffectCount = new java.util.concurrent.atomic.AtomicInteger(0)
+    val sut = Observable.create(AsyncOnSubscribe(() => sideEffectCount.incrementAndGet())((s, _) => (Notification.OnCompleted, s)))
+    sut.toBlocking.toList
+    sut.toBlocking.toList
+    assertEquals(sideEffectCount.get(), 2)
+  }
+}

src/test/scala/rx/lang/scala/observables/SyncOnSubscribeTests.scala

@@ -0,0 +1,61 @@
+package rx.lang.scala.observables
+
+import org.junit.Test
+import org.junit.Assert._
+import org.scalatest.junit.JUnitSuite
+import rx.lang.scala.observers.TestSubscriber
+import rx.lang.scala.{Notification, Observable}
+
+class SyncOnSubscribeTests extends JUnitSuite {
+
+  @Test
+  def testStateful(): Unit = {
+    val sut = Observable.create(SyncOnSubscribe(() => 0)(count =>
+      if(count > 3)
+        (Notification.OnCompleted, count)
+      else
+        (Notification.OnNext(count), count+1)
+    ))
+    assertEquals(List(0,1,2,3), sut.toBlocking.toList)
+  }
+
+  @Test
+  def testStateless(): Unit = {
+    val sut = Observable.create(SyncOnSubscribe.stateless(() => Notification.OnNext(42)))
+    assertEquals(List(42,42,42,42), sut.take(4).toBlocking.toList)
+  }
+
+  @Test
+  def testSingleState(): Unit = {
+    val random = math.random
+    val sut = Observable.create(SyncOnSubscribe.singleState(() => random)(s => Notification.OnNext(s.toString)))
+    assertEquals(List(random.toString, random.toString), sut.take(2).toBlocking.toList)
+  }
+
+  @Test
+  def testUnsubscribe(): Unit = {
+    val sideEffect = new java.util.concurrent.atomic.AtomicBoolean(false)
+    val sut = Observable.create(SyncOnSubscribe(() => ())(s => (Notification.OnCompleted, s), onUnsubscribe = s => sideEffect.set(true)))
+    sut.foreach(_ => ())
+    assertEquals(true, sideEffect.get())
+  }
+
+  @Test
+  def testError(): Unit = {
+    val e = new IllegalStateException("Oh noes")
+    val sut = Observable.create(SyncOnSubscribe(() => 0)(s => (if(s>2) Notification.OnNext(s) else Notification.OnError(e), s+1)))
+    val testSubscriber = TestSubscriber[Int]()
+    sut.subscribe(testSubscriber)
+    testSubscriber.assertError(e)
+  }
+
+  @Test
+  // Ensure that the generator is executed for each subscription
+  def testGenerator(): Unit = {
+    val sideEffectCount = new java.util.concurrent.atomic.AtomicInteger(0)
+    val sut = Observable.create(SyncOnSubscribe(() => sideEffectCount.incrementAndGet())(s => (Notification.OnCompleted, s)))
+    sut.toBlocking.toList
+    sut.toBlocking.toList
+    assertEquals(sideEffectCount.get(), 2)
+  }
+}