Cleaned up subscriptions to make them idempotent when called from either side.

Author: AppliedDuality

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

@@ -180,17 +180,6 @@ trait Scheduler {
         work{ t1.call() }
       }
     }))
-    //action1[action0]
-
-//    val subscription = new rx.subscriptions.MultipleAssignmentSubscription()
-//
-//    subscription.setSubscription(
-//      this.schedule(scheduler => {
-//        def loop(): Unit =  subscription.setSubscription(scheduler.schedule{ work{ loop() }})
-//        loop()
-//        subscription
-//      }))
-//    subscription
   }
 
   /**

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

@@ -26,28 +26,24 @@ import java.util.concurrent.atomic.AtomicBoolean
  */
 trait Subscription {
 
-  private [scala] val asJavaSubscription: rx.Subscription = new rx.Subscription {
-    override def unsubscribe(){  Subscription.this.unsubscribe() }
-  }
-
+  private [scala] val asJavaSubscription: rx.Subscription
+  private [scala] val unsubscribed = new AtomicBoolean(false)
 
   /**
    * Call this method to stop receiving notifications on the Observer that was registered when
    * this Subscription was received.
    */
-  def unsubscribe(): Unit = { unsubscribed.set(true) }
+  def unsubscribe() = asJavaSubscription.unsubscribe()
 
   /**
    * Checks if the subscription is unsubscribed.
    */
-  def isUnsubscribed: Boolean = unsubscribed.get()
-  private [scala] val unsubscribed = new AtomicBoolean(false)
+  def isUnsubscribed = unsubscribed.get()
 
 }
 
 object Subscription {
 
-
   /**
    * Creates an [[rx.lang.scala.Subscription]] from an [[rx.Subscription]].             ß
    */
@@ -57,24 +53,22 @@ object Subscription {
       case x: rx.subscriptions.CompositeSubscription => new rx.lang.scala.subscriptions.CompositeSubscription(x)
       case x: rx.subscriptions.MultipleAssignmentSubscription => new rx.lang.scala.subscriptions.MultipleAssignmentSubscription(x)
       case x: rx.subscriptions.SerialSubscription => new rx.lang.scala.subscriptions.SerialSubscription(x)
-      case x: rx.Subscription => apply { x.unsubscribe() }
+      case x: rx.Subscription => apply{ x.unsubscribe }
     }
   }
 
   /**
    * Creates an [[rx.lang.scala.Subscription]] that invokes the specified action when unsubscribed.
    */
-  def apply(u: => Unit): Subscription =  {
-    new Subscription() {
-      override def unsubscribe(): Unit = {
-        if(!super.isUnsubscribed) { u; super.unsubscribe() }
-      }
+  def apply(u: => Unit): Subscription = new Subscription() {
+    val asJavaSubscription = new rx.Subscription {
+      override def unsubscribe() { if(unsubscribed.compareAndSet(false, true)) { u } }
     }
   }
 
   /**
-   * Checks if the subscription is unsubscribed.
+   * Creates an empty [[rx.lang.scala.Subscription]].
    */
-  def apply(): Subscription = { new Subscription {} }
+  def apply(): Subscription = Subscription {}
 
 }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/subscriptions/BooleanSubscription.scala

@@ -17,17 +17,22 @@ package rx.lang.scala.subscriptions
 
 import rx.lang.scala._
 
+private [scala] object BooleanSubscription {
+  def apply(): BooleanSubscription = new BooleanSubscription(new rx.subscriptions.BooleanSubscription())
+}
+
 /**
  * Represents a [[rx.lang.scala.Subscription]] that can be checked for status.
  */
-private [scala] class BooleanSubscription private[scala] (override val asJavaSubscription: rx.subscriptions.BooleanSubscription)
+private [scala] class BooleanSubscription private[scala] (boolean: rx.subscriptions.BooleanSubscription)
   extends Subscription {
 
-  //override def asJavaSubscription =  subscription
-
-  /**
-   * Checks whether the subscription has been unsubscribed.
-   */
-  override def isUnsubscribed: Boolean = asJavaSubscription.isUnsubscribed
-  override def unsubscribe(): Unit =  asJavaSubscription.unsubscribe()
+  val asJavaSubscription: rx.subscriptions.BooleanSubscription = new rx.subscriptions.BooleanSubscription() {
+    override def unsubscribe(): Unit = {
+      if(unsubscribed.compareAndSet(false, true)) {
+        if(!boolean.isUnsubscribed) { boolean.unsubscribe() }
+      }
+    }
+    override def isUnsubscribed(): Boolean = unsubscribed.get() || boolean.isUnsubscribed
+  }
 }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/subscriptions/SerialSubscription.scala

@@ -17,38 +17,39 @@ package rx.lang.scala.subscriptions
 
 import rx.lang.scala._
 
-
 object SerialSubscription {
 
   /**
    * Creates a [[rx.lang.scala.subscriptions.SerialSubscription]].
    */
-  def apply(): SerialSubscription =  {
-    new SerialSubscription(new rx.subscriptions.SerialSubscription())
-  }
+  def apply(): SerialSubscription =  new SerialSubscription(new rx.subscriptions.SerialSubscription())
 
   /**
    * Creates a [[rx.lang.scala.subscriptions.SerialSubscription]] that invokes the specified action when unsubscribed.
    */
   def apply(unsubscribe: => Unit): SerialSubscription = {
-    val s= SerialSubscription()
-    s.subscription  = Subscription{ unsubscribe }
-    s
+   SerialSubscription().subscription = Subscription(unsubscribe)
   }
 }
 
 /**
  * Represents a [[rx.lang.scala.Subscription]] that can be checked for status.
  */
-class SerialSubscription private[scala] (override val asJavaSubscription: rx.subscriptions.SerialSubscription) extends Subscription {
+class SerialSubscription private[scala] (serial: rx.subscriptions.SerialSubscription) extends Subscription {
 
-  //override def asJavaSubscription = s
-  /**
-   * Unsubscribes this subscription, setting isUnsubscribed to true.
+  /*
+  * As long as rx.subscriptions.SerialSubscription has no isUnsubscribed,
+  * we need to intercept and do it ourselves.
    */
-  override def unsubscribe(): Unit = { asJavaSubscription.unsubscribe(); unsubscribed.set(true) }
+  val asJavaSubscription: rx.subscriptions.SerialSubscription = new rx.subscriptions.SerialSubscription() {
+    override def unsubscribe(): Unit = {
+      if(unsubscribed.compareAndSet(false, true)) { serial.unsubscribe() }
+    }
+    override def setSubscription(subscription: rx.Subscription): Unit = serial.setSubscription(subscription)
+    override def getSubscription(): rx.Subscription = serial.getSubscription()
+  }
 
-  def subscription_=(value: Subscription): Unit = asJavaSubscription.setSubscription(value.asJavaSubscription)
+  def subscription_=(value: Subscription): this.type = { asJavaSubscription.setSubscription(value.asJavaSubscription); this }
   def subscription: Subscription = Subscription(asJavaSubscription.getSubscription)
 
 }

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

@@ -24,11 +24,13 @@ class SubscriptionTests extends JUnitSuite {
       assertFalse(subscription.isUnsubscribed)
 
     subscription.unsubscribe()
+
       assertTrue(subscription.isUnsubscribed)
   }
 
   @Test
   def subscriptionUnsubscribeIdempotent() {
+
     var called = false
 
     val subscription = Subscription{ called = !called }
@@ -37,10 +39,12 @@ class SubscriptionTests extends JUnitSuite {
       assertFalse(subscription.isUnsubscribed)
 
     subscription.unsubscribe()
+
       assertTrue(called)
       assertTrue(subscription.isUnsubscribed)
 
     subscription.unsubscribe()
+
       assertTrue(called)
       assertTrue(subscription.isUnsubscribed)
   }
@@ -62,11 +66,14 @@ class SubscriptionTests extends JUnitSuite {
 
       assertTrue(composite.isUnsubscribed)
       assertTrue(s0.isUnsubscribed)
-      assertTrue(s0.isUnsubscribed)
+      assertTrue(s1.isUnsubscribed)
 
     val s2 = Subscription{}
+
       assertFalse(s2.isUnsubscribed)
+
     composite += s2
+
       assertTrue(s2.isUnsubscribed)
 
   }
@@ -84,7 +91,9 @@ class SubscriptionTests extends JUnitSuite {
       assertTrue(s0.isUnsubscribed)
 
     composite.unsubscribe()
+
       assertTrue(composite.isUnsubscribed)
+      assertTrue(s0.isUnsubscribed)
   }
 
   @Test
@@ -99,21 +108,27 @@ class SubscriptionTests extends JUnitSuite {
         assertFalse(s1.isUnsubscribed)
 
       multiple.subscription = s0
+
         assertFalse(s0.isUnsubscribed)
         assertFalse(s1.isUnsubscribed)
 
       multiple.subscription = s1
+
         assertFalse(s0.isUnsubscribed)   // difference with SerialSubscription
         assertFalse(s1.isUnsubscribed)
 
       multiple.unsubscribe()
+
         assertTrue(multiple.isUnsubscribed)
         assertFalse(s0.isUnsubscribed)
         assertTrue(s1.isUnsubscribed)
 
       val s2 = Subscription()
+
         assertFalse(s2.isUnsubscribed)
+
       multiple.subscription = s2
+
         assertTrue(s2.isUnsubscribed)
         assertFalse(s0.isUnsubscribed)
   }
@@ -130,20 +145,27 @@ class SubscriptionTests extends JUnitSuite {
       assertFalse(s1.isUnsubscribed)
 
     serial.subscription = s0
+
       assertFalse(s0.isUnsubscribed)
       assertFalse(s1.isUnsubscribed)
 
     serial.subscription = s1
+
       assertTrue(s0.isUnsubscribed)    // difference with MultipleAssignmentSubscription
       assertFalse(s1.isUnsubscribed)
 
     serial.unsubscribe()
+
       assertTrue(serial.isUnsubscribed)
       assertTrue(s1.isUnsubscribed)
 
     val s2 = Subscription()
+
       assertFalse(s2.isUnsubscribed)
+
     serial.subscription = s2
+
       assertTrue(s2.isUnsubscribed)
   }
+
 }