RxScala: Fix recursive onStart call in Subscriber

Author: jbripley

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

@@ -5,7 +5,6 @@ trait Subscriber[-T] extends Observer[T] with Subscription {
   self =>
 
   private [scala] val asJavaSubscriber: rx.Subscriber[_ >: T] = new rx.Subscriber[T] with SubscriberAdapter[T] {
-    override def onStart(): Unit = self.onStart()
     override def onNext(value: T): Unit = self.onNext(value)
     override def onError(error: Throwable): Unit = self.onError(error)
     override def onCompleted(): Unit = self.onCompleted()
@@ -36,7 +35,7 @@ trait Subscriber[-T] extends Observer[T] with Subscription {
   }
 
   override final def isUnsubscribed: Boolean = {
-    asJavaSubscriber.isUnsubscribed()
+    asJavaSubscriber.isUnsubscribed
   }
 
   def onStart(): Unit = {

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

@@ -18,6 +18,8 @@ package rx.lang.scala
 import org.junit.Test
 import org.junit.Assert.assertNotNull
 import org.junit.Assert.assertTrue
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertEquals
 import org.scalatest.junit.JUnitSuite
 
 class SubscriberTests extends JUnitSuite {
@@ -57,4 +59,27 @@ class SubscriberTests extends JUnitSuite {
     assertTrue(subscription.isUnsubscribed)
   }
 
+  @Test def testNewSubscriber(): Unit = {
+    var didComplete = false
+    var didError = false
+    var onNextValue = 0
+
+    Observable.just(1).subscribe(new Subscriber[Int] {
+      override def onCompleted(): Unit = {
+        didComplete = true
+      }
+
+      override def onError(e: Throwable): Unit = {
+        didError = true
+      }
+
+      override def onNext(v: Int): Unit = {
+        onNextValue = v
+      }
+    })
+
+    assertTrue("Subscriber called onCompleted", didComplete)
+    assertFalse("Subscriber did not call onError", didError)
+    assertEquals(1, onNextValue)
+  }
 }