Merge pull request #1362 from zsxwing/rxscala-create

RxScala: Fix #1340 and #1343

Author: benjchristensen

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

@@ -4018,13 +4018,15 @@ object Observable {
    * @return
    *         an Observable that, when an [[rx.lang.scala.Observer]] subscribes to it, will execute the given function.
    */
-  @deprecated("Use `apply[T](Subscriber[T] => Unit)` instead", "0.17.0")
   def create[T](func: Observer[T] => Subscription): Observable[T] = {
-    toScalaObservable[T](rx.Observable.create(new OnSubscribeFunc[T] {
-      def onSubscribe(t1: rx.Observer[_ >: T]): rx.Subscription = {
-        func(Observer(t1))
+    Observable(
+      (subscriber: Subscriber[T]) => {
+        val s = func(subscriber)
+        if (s != null && s != subscriber) {
+          subscriber.add(s)
+        }
       }
-    }))
+    )
   }
 
   /*
@@ -4305,7 +4307,6 @@ object Observable {
    */
   def interval(duration: Duration): Observable[Long] = {
     toScalaObservable[java.lang.Long](rx.Observable.interval(duration.length, duration.unit)).map(_.longValue())
-    /*XXX*/
   }
 
   /**
@@ -4338,7 +4339,6 @@ object Observable {
    */
   def timer(initialDelay: Duration, period: Duration): Observable[Long] = {
     toScalaObservable[java.lang.Long](rx.Observable.timer(initialDelay.toNanos, period.toNanos, duration.NANOSECONDS)).map(_.longValue())
-    /*XXX*/
   }
 
   /**

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/ObservableTest.scala

@@ -271,4 +271,19 @@ class ObservableTests extends JUnitSuite {
     assertEquals(expected, r)
   }
 
+  @Test
+  def testCreate() {
+    var called = false
+    val o = Observable.create[String](observer => {
+      observer.onNext("a")
+      observer.onNext("b")
+      observer.onNext("c")
+      observer.onCompleted()
+      Subscription {
+        called = true
+      }
+    })
+    assertEquals(List("a", "b", "c"), o.toBlocking.toList)
+    assertTrue(called)
+  }
 }