RxScala: Add mapping to RxJava doOnUnsubscribe

Author: jbripley

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

@@ -843,6 +843,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

@@ -3608,6 +3608,26 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.doOnTerminate(() => onTerminate))
   }
 
+  /**
+   * Modifies the source {@code 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 {@code Observable} is reference counted, in which case the source {@code 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>{@code doOnUnsubscribe} does not operate by default on a particular {@link Scheduler}.</dd>
+   * </dl>
+   *
+   * @param onUnsubscribe the action that gets called when this { @code Observable} is unsubscribed
+   * @return the source { @code 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>
+   */
+  def doOnUnsubscribe(onUnsubscribe: => Unit): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.doOnUnsubscribe(() => onUnsubscribe))
+  }
+
   /**
    * Given two Observables, mirror the one that first emits an item.
    *