Merge pull request #1598 from jbripley/rxscala-doonunsubscribe

RxScala: Add mapping to RxJava doOnUnsubscribe

Author: benjchristensen

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

@@ -833,6 +833,16 @@ class RxScalaDemo extends JUnitSuite {
     // onCompleted
   }
 
+  @Test def doOnSubscribeExample(): Unit = {
+    val o = List("red", "green", "blue").toObservable.doOnSubscribe { println("subscribed") }
+    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
+    // subscribed
+    // red
+    // green
+    // blue
+    // onCompleted
+  }
+
   @Test def doOnTerminateExample(): Unit = {
     val o = List("red", "green", "blue").toObservable.doOnTerminate { println("terminate") }
     o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
@@ -843,6 +853,16 @@ class RxScalaDemo extends JUnitSuite {
     // onCompleted
   }
 
+  @Test def doOnUnsubscribeExample(): Unit = {
+    val o = List("red", "green", "blue").toObservable.doOnUnsubscribe { println("unsubscribed") }
+    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
+    // red
+    // green
+    // blue
+    // onCompleted
+    // unsubscribed
+  }
+
   @Test def finallyDoExample(): Unit = {
     val o = List("red", "green", "blue").toObservable.finallyDo { println("finally") }
     o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))

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

@@ -3592,6 +3592,28 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.doOnEach(Observer(onNext, onError,onCompleted)))
   }
 
+  /**
+   * Modifies the source `Observable` so that it invokes the given action when it is subscribed from
+   * its subscribers. Each subscription will result in an invocation of the given action except when the
+   * source `Observable` is reference counted, in which case the source `Observable` will invoke
+   * the given action for the first subscription.
+   * <p>
+   * <img width="640" height="390" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/doOnSubscribe.png" alt="">
+   * <dl>
+   *  <dt><b>Scheduler:</b></dt>
+   *  <dd>`onSubscribe` does not operate by default on a particular `Scheduler`.</dd>
+   * </dl>
+   *
+   * @param onSubscribe
+     *            the action that gets called when an observer subscribes to this `Observable`
+   * @return the source `Observable` modified so as to call this Action when appropriate
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#doonsubscribe">RxJava wiki: doOnSubscribe</a>
+   * @since 0.20
+   */
+  def doOnSubscribe(onSubscribe: => Unit): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.doOnSubscribe(() => onSubscribe))
+  }
+
   /**
    * Modifies an Observable so that it invokes an action when it calls `onCompleted` or `onError`.
    * <p>
@@ -3608,6 +3630,28 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.doOnTerminate(() => onTerminate))
   }
 
+  /**
+   * Modifies the source `Observable` so that it invokes the given action when it is unsubscribed from
+   * its subscribers. Each un-subscription will result in an invocation of the given action except when the
+   * source `Observable` is reference counted, in which case the source `Observable` will invoke
+   * the given action for the very last un-subscription.
+   * <p>
+   * <img width="640" height="310" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/doOnUnsubscribe.png" alt="">
+   * <dl>
+   *  <dt><b>Scheduler:</b></dt>
+   *  <dd>`doOnUnsubscribe` does not operate by default on a particular `Scheduler`.</dd>
+   * </dl>
+   *
+   * @param onUnsubscribe
+     *            the action that gets called when this `Observable` is unsubscribed
+   * @return the source `Observable` modified so as to call this Action when appropriate
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#doonunsubscribe">RxJava wiki: doOnUnsubscribe</a>
+   * @since 0.20
+   */
+  def doOnUnsubscribe(onUnsubscribe: => Unit): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.doOnUnsubscribe(() => onUnsubscribe))
+  }
+
   /**
    * Given two Observables, mirror the one that first emits an item.
    *