Remove global subscription store in place of per-isolate stores.

Author: axel22

src/main/scala/scala/reactive/EventSink.scala

@@ -2,7 +2,6 @@ package scala.reactive
 
 
 
-import scala.collection._
 
 
 
@@ -12,27 +11,17 @@ import scala.collection._
  */
 trait EventSink {
 
-  def registerEventSink() {
-    Iso.selfIso.get match {
-      case null =>
-        EventSink.globalEventSinks += this
-      case iso =>
-        iso.eventSinks += this
-    }
+  def registerEventSink(canLeak: CanLeak) {
+    canLeak.eventSinks += this
   }
 
-  def unregisterEventSink() {
-    Iso.selfIso.get match {
-      case null =>
-        EventSink.globalEventSinks -= this
-      case iso =>
-        iso.eventSinks -= this
-    }
+  def unregisterEventSink(canLeak: CanLeak) {
+    canLeak.eventSinks -= this
   }
 
-  def liftSubscription(s: Reactive.Subscription) = {
+  def liftSubscription(s: Reactive.Subscription, canLeak: CanLeak) = {
     Reactive.Subscription {
-      unregisterEventSink()
+      unregisterEventSink(canLeak)
       s.unsubscribe()
     }
   }
@@ -42,7 +31,4 @@ trait EventSink {
 
 object EventSink {
 
-  /** A set of global event sinks. */
-  private[reactive] val globalEventSinks = mutable.Set[EventSink]()
-
 }

src/main/scala/scala/reactive/Iso.scala

@@ -71,14 +71,12 @@ trait Iso[@spec(Int, Long, Double) T] extends ReactRecord {
 
   @volatile private[reactive] var eventSources: mutable.Set[EventSource] = _
 
-  @volatile private[reactive] var eventSinks: mutable.Set[EventSink] = _
-
   @volatile private[reactive] var systemEmitter: Reactive.Emitter[SysEvent] = _
 
   @volatile private[reactive] var failureEmitter:
     Reactive.Emitter[Throwable] = _
 
-  val implicits = Iso.Implicits
+  val implicits = new Iso.Implicits
 
   private def illegal() =
     throw new IllegalStateException("Only isolate systems can create isolates.")
@@ -91,7 +89,6 @@ trait Iso[@spec(Int, Long, Double) T] extends ReactRecord {
       case eq => eq.asInstanceOf[IsoFrame]
     }
     eventSources = mutable.Set[EventSource]()
-    eventSinks = mutable.Set[EventSink]()
     systemEmitter = new Reactive.Emitter[SysEvent]
     failureEmitter = new Reactive.Emitter[Throwable]
 
@@ -191,8 +188,8 @@ object Iso {
    */
   def of[@specialized(Int, Long, Double) T]: Iso[T] = Iso.self[Iso[T]]
 
-  object Implicits {
-    implicit def canLeak: CanLeak = CanLeak.isoCanLeak
+  class Implicits {
+    implicit val canLeak: CanLeak = CanLeak.newCanLeak
   }
 
 }

src/main/scala/scala/reactive/Reactive.scala

@@ -83,9 +83,9 @@ trait Reactive[@spec(Int, Long, Double) +T] {
    */
   def onReaction(reactor: Reactor[T])(implicit canLeak: CanLeak):
     Reactive.Subscription = {
-    val eventSink = new Reactor.EventSink(reactor)
+    val eventSink = new Reactor.EventSink(reactor, canLeak)
     val subscription = observe(eventSink)
-    eventSink.liftSubscription(subscription)
+    eventSink.liftSubscription(subscription, canLeak)
   }
 
   /** A shorthand for `onReaction` -- the specified functions are invoked

src/main/scala/scala/reactive/Reactor.scala

@@ -28,18 +28,19 @@ trait Reactor[@spec(Int, Long, Double) -T] {
 
 object Reactor {
 
-  class EventSink[@spec(Int, Long, Double) T](val underlying: Reactor[T])
+  class EventSink[@spec(Int, Long, Double) T]
+    (val underlying: Reactor[T], val canLeak: CanLeak)
   extends Reactor[T] with scala.reactive.EventSink {
     def init(dummy: EventSink[T]) {
-      registerEventSink()
+      registerEventSink(canLeak)
     }
 
     init(this)
 
     def react(value: T) = underlying.react(value)
 
     def unreact() = {
-      unregisterEventSink()
+      unregisterEventSink(canLeak)
       underlying.unreact()
     }
   }

src/main/scala/scala/reactive/package.scala

@@ -3,6 +3,7 @@ package scala
 
 
 import scala.annotation.implicitNotFound
+import scala.collection._
 import scala.reflect.ClassTag
 
 
@@ -248,11 +249,12 @@ package object reactive {
     "If you are sure you want to risk this, import " +
     "scala.reactive.Implicits.canLeak. " +
     "Otherwise, consider calling observe or foreach.")
-  sealed trait CanLeak
+  sealed trait CanLeak {
+    val eventSinks = mutable.Set[EventSink]()
+  }
 
   object CanLeak {
     private[reactive] def newCanLeak: CanLeak = new CanLeak {}
-    private[reactive] val isoCanLeak: CanLeak = new CanLeak {}
   }
 
   /* system events */

src/test/scala/scala/reactive/test/signal/ReactiveGCSpec.scala

@@ -57,12 +57,8 @@ class ReactiveGCSpec extends FlatSpec with ShouldMatchers {
 
     emitter += 1
 
-    try {
-      for (i <- 0 until num) signsOfLife(i) should equal (true)
-      afterCheck(emitter) should equal (true)
-    } finally {
-      EventSink.globalEventSinks.clear()
-    }
+    for (i <- 0 until num) signsOfLife(i) should equal (true)
+    afterCheck(emitter) should equal (true)
   }
 
   it should "not GC 1 onX dependency" in {