Add unsafeSubscribe to RxScala

Author: zsxwing

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

@@ -148,6 +148,22 @@ trait Observable[+T]
     thisJava.subscribe(subscriber.asJavaSubscriber)
   }
 
+  /**
+   * Subscribe to Observable and invoke `OnSubscribe` function without any
+   * contract protection, error handling, unsubscribe, or execution hooks.
+   *
+   * This should only be used for implementing an `Operator` that requires nested subscriptions.
+   *
+   * Normal use should use [[Observable.subscribe]] which ensures the Rx contract and other functionality.
+   *
+   * @param subscriber
+   * @return [[Subscription]] which is the Subscriber passed in
+   * @since 0.17
+   */
+  def unsafeSubscribe(subscriber: Subscriber[T]): Subscription = {
+    asJavaObservable.unsafeSubscribe(subscriber.asJavaSubscriber)
+  }
+
   /**
    * $subscribeSubscriberMain
    *