Add 'doOnTerminate' to RxScala

Author: zsxwing

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

@@ -560,6 +560,26 @@ class RxScalaDemo extends JUnitSuite {
     obs.toBlockingObservable.toIterable.last
   }
 
+  @Test def doOnTerminateExample(): Unit = {
+    val o = List("red", "green", "blue").toObservable.doOnTerminate(() => println("terminate"))
+    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
+    // red
+    // green
+    // blud
+    // terminate
+    // onCompleted
+  }
+
+  @Test def finallyDoExample(): Unit = {
+    val o = List("red", "green", "blue").toObservable.finallyDo(() => println("finally"))
+    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
+    // red
+    // green
+    // blud
+    // onCompleted
+    // finally
+  }
+
   @Test def timeoutExample(): Unit = {
     val other = List(100L, 200L, 300L).toObservable
     val result = Observable.interval(100 millis).timeout(50 millis, other).toBlockingObservable.toList

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

@@ -2461,6 +2461,22 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.doOnEach(Observer(onNext, onError,onCompleted)))
   }
 
+  /**
+   * Modifies an Observable so that it invokes an action when it calls `onCompleted` or `onError`.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/doOnTerminate.png">
+   * <p>
+   * This differs from `finallyDo` in that this happens BEFORE onCompleted/onError are emitted.
+   *
+   * @param onTerminate the action to invoke when the source Observable calls `onCompleted` or `onError`
+   * @return the source Observable with the side-effecting behavior applied
+   * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-doonterminate">RxJava Wiki: doOnTerminate()</a>
+   * @see <a href="http://msdn.microsoft.com/en-us/library/hh229804.aspx">MSDN: Observable.Do</a>
+   */
+  def doOnTerminate(onTerminate: () => Unit): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.doOnTerminate(onTerminate))
+  }
+
   /**
    * Given two Observables, mirror the one that first emits an item.
    *