Add unsubscribeOn to RxScala

Author: zsxwing

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

@@ -1048,4 +1048,16 @@ class RxScalaDemo extends JUnitSuite {
     val o = hot.publishLast(co => co ++ co) // "++" subscribes "co" twice
     o.subscribe(n => println(s"subscriber gets $n"))
   }
+
+  @Test def unsubscribeOnExample() {
+    val o = Observable[String] {
+      subscriber =>
+        subscriber.add(Subscription {
+          println("unsubscribe on " + Thread.currentThread().getName())
+        })
+        subscriber.onNext("RxScala")
+        subscriber.onCompleted()
+    }
+    o.unsubscribeOn(NewThreadScheduler()).subscribe(println(_))
+  }
 }

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

@@ -949,6 +949,17 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.subscribeOn(scheduler))
   }
 
+  /**
+   * Asynchronously unsubscribes on the specified [[Scheduler]].
+   *
+   * @param scheduler the [[Scheduler]] to perform subscription and unsubscription actions on
+   * @return the source Observable modified so that its unsubscriptions happen on the specified [[Scheduler]]
+   * @since 0.17
+   */
+  def unsubscribeOn(scheduler: Scheduler): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.unsubscribeOn(scheduler))
+  }
+
   /**
    * Asynchronously notify [[rx.lang.scala.Observer]]s on the specified [[rx.lang.scala.Scheduler]].
    *