Automatic channel registration and unregistration in the iso system.

Author: Cantor Chron

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

@@ -96,26 +96,6 @@ trait Iso[@spec(Int, Long, Double) T] extends ReactRecord {
    */
   react <<= frame.internalConnector.events.collect({ case e: SysEvent => e }).pipe(systemEmitter)
 
-  /** Opens a new channel for this isolate.
-   *
-   *  @tparam Q        type of the events in the new channel
-   *  @param factory   event queue factory
-   *  @return          the connector object of the new channel
-   */
-  final def open[@spec(Int, Long, Double) Q: Arrayable](implicit factory: EventQueue.Factory = frame.eventQueueFactory): Connector[Q] =
-    Iso.openChannel[Q](frame, factory, false)
-
-  /** Opens a new daemon channel for this isolate.
-   *
-   *  Daemon channels do not affect isolate termination.
-   *
-   *  @tparam Q        type of the events in the new channel
-   *  @param factory   event queue factory
-   *  @return          the connector object of the new channel
-   */
-  final def daemon[@spec(Int, Long, Double) Q: Arrayable](implicit factory: EventQueue.Factory = frame.eventQueueFactory): Connector[Q] =
-    Iso.openChannel[Q](frame, factory, true)
-
   /** The unique id of this isolate.
    *  
    *  @return          the unique id, assigned only to this isolate
@@ -165,13 +145,6 @@ object Iso {
 
   private[reactive] val argFrame = new DynamicVariable[IsoFrame](null)
 
-  private[reactive] def openChannel[@spec(Int, Long, Double) Q: Arrayable](frame: IsoFrame, factory: EventQueue.Factory, isDaemon: Boolean): Connector[Q] = {
-    val eventQueue = factory.create[Q]
-    val connector = new Connector(frame, eventQueue, isDaemon)
-    frame.multiplexer += connector
-    connector
-  }
-
   /** Returns the current isolate.
    *
    *  If the caller is not executing in an isolate,

src/main/scala/scala/reactive/IsoSystem.scala

@@ -54,7 +54,7 @@ abstract class IsoSystem {
    *  @param scheduler  the scheduler used to scheduler the isolate
    *  @return           the channel for this isolate
    */
-  def isolate[@spec(Int, Long, Double) T: Arrayable](proto: Proto[Iso[T]], name: String = null): Channel[T]
+  def isolate[@spec(Int, Long, Double) T: Arrayable](proto: Proto[Iso[T]]): Channel[T]
 
   /** Creates a new channel for the specified isolate frame.
    *
@@ -77,11 +77,14 @@ abstract class IsoSystem {
    */
   protected def uniqueName(name: String): String
 
-  /** Releases the name after the isolate terminates.
+  /** Releases the channel names associated with the isolate,
+   *  and then releases the name of the isolate.
+   *
+   *  Called after the isolate terminates.
    *  
    *  @param name       the name to release
    */
-  protected[reactive] def releaseName(name: String): Unit
+  protected[reactive] def releaseNames(name: String): Unit
 
   /** Generates a new unique id, generated only once during
    *  the lifetime of this isolate system.
@@ -107,14 +110,18 @@ abstract class IsoSystem {
   /** Creates an isolate frame.
    *
    *  Should only be overridden if the default isolate initialization order needs to change.
+   *  The multiplexer, unique name and unique id are created for an isolate first.
+   *  Then, the isolate frame is created.
+   *  Then, the isolate object (concrete user implementation) is instantiated.
+   *  Then, the isolate frame is assigned the isolate object.
+   *  Finally, the `initiate` method is called on the scheduler, and the isolate is returned.
    *  See the source code of the default implementation of this method for more details.
    *
    *  @tparam T         the type of the events for the isolate
    *  @param proto      prototype for the isolate
-   *  @param name       name of the new isolate
    *  @return           the resulting isolate frame
    */
-  protected def createFrame[@spec(Int, Long, Double) T: Arrayable](proto: Proto[Iso[T]], name: String): Iso[T] = {
+  protected def createFrame[@spec(Int, Long, Double) T: Arrayable](proto: Proto[Iso[T]]): Iso[T] = {
     val scheduler = proto.scheduler match {
       case null => bundle.defaultScheduler
       case name => bundle.scheduler(name)
@@ -128,16 +135,16 @@ abstract class IsoSystem {
       case mult => mult
     }
     val uid = uniqueId()
-    val uname = uniqueName(name)
+    val uname = uniqueName(proto.name)
     val frame = new IsoFrame(
       uid,
       uname,
       IsoSystem.this,
       scheduler,
       queueFactory,
       multiplexer,
-      frame => Iso.openChannel(frame, queueFactory, false),
-      frame => Iso.openChannel(frame, queueFactory, true)
+      frame => IsoSystem.openChannel(IsoSystem.this, frame, queueFactory, "events", false),
+      frame => IsoSystem.openChannel(IsoSystem.this, frame, queueFactory, "internal", true)
     )
     val isolate = Iso.argFrame.withValue(frame) {
       createAndResetIso(proto)
@@ -154,6 +161,19 @@ abstract class IsoSystem {
  */
 object IsoSystem {
 
+  /** Opens a channel for the current isolate, using the specified parameters.
+   */
+  private[reactive] def openChannel[@spec(Int, Long, Double) Q: Arrayable](system: IsoSystem, frame: IsoFrame, f: EventQueue.Factory, cn: String, isDaemon: Boolean): Connector[Q] = {
+    val factory = if (f != null) f else Iso.self.frame.eventQueueFactory
+    val channelName = if (cn != null) cn else "channel-" + frame.counter.incrementAndGet().toString
+    val eventQueue = factory.create[Q]
+    val name = frame.name + "#" + channelName
+    val connector = new Connector(frame, eventQueue, name, isDaemon)
+    system.channels(name) = connector.channel
+    frame.multiplexer += connector
+    connector
+  }
+
   /** Retrieves the default isolate system.
    *  
    *  @param name       the name for the isolate system instance
@@ -210,41 +230,58 @@ object IsoSystem {
    */
   lazy val defaultBundle = Bundle.default(Scheduler.default)
 
+  class ChannelBuilder(
+    val system: IsoSystem,
+    val channelName: String,
+    val isDaemon: Boolean,
+    val eventQueueFactory: EventQueue.Factory
+  ) {
+    /** Associates a new name for the channel.
+     */
+    def named(name: String) = new ChannelBuilder(system, name, isDaemon, eventQueueFactory)
+
+    /** Specifies a daemon channel.
+     */
+    def daemon = new ChannelBuilder(system, channelName, true, eventQueueFactory)
+
+    /** Associates a new event queue factory.
+     */
+    def eventQueue(factory: EventQueue.Factory) = new ChannelBuilder(system, channelName, isDaemon, factory)
+
+    /** Opens a new channel for this isolate.
+     *
+     *  @tparam Q        type of the events in the new channel
+     *  @param factory   event queue factory
+     *  @return          the connector object of the new channel
+     */
+    final def open[@spec(Int, Long, Double) Q: Arrayable]: Connector[Q] =
+      IsoSystem.openChannel[Q](system, Iso.self.frame, eventQueueFactory, channelName, isDaemon)
+  }
+
   /** The channel register used for channel lookup by name.
    */
-  trait Channels {
+  abstract class Channels(system: IsoSystem) extends ChannelBuilder(system, null, false, null) {
+
     /** Registers a new channel with this isolate system.
      *
      *  Throws an exception if name is already taken.
      *  
      *  @param name       name of the channel
      *  @param channel    the channel to register
      */
-    def update(name: String, channel: Channel[_]): Unit
+    private[reactive] def update(name: String, channel: Channel[_]): Unit
 
     /** Removes the channel registration.
      *
      *  @param name       name of the channel to remove from the register
      */
-    def remove(name: String): Unit
-
-    /** Returns a channel under the specified name, if any.
-     *
-     *  Throws an exception if such a channel does not exist.
-     *  
-     *  @param name       name of the channel
-     *  @return           the channel registered under the specified name
-     */
-    def apply[@spec(Int, Long, Double) T](name: String): Channel[T]
+    private[reactive] def remove(name: String): Unit
 
-    /** Returns a channel under the specified name, if any.
+    /** Removes all the channels registered with a specific isolate.
      *
-     *  Returns `None` if none exist.
-     *  
-     *  @param name       name of the channel
-     *  @return           optionally, the channel registered under the specified name
+     *  @param isoName    name of the isolate whose channels must be removed
      */
-    def get[T](name: String): Option[Channel[T]]
+    private[reactive] def removeIsolate(isoName: String): Unit
 
     /** Eventually returns a channel under the specified name.
      *
@@ -255,6 +292,9 @@ object IsoSystem {
 
     /** Eventually returns an *unsealed* channel under the specified name.
      *
+     *  Note: between the time that the channel is retrieved and the time it is first used,
+     *  that channel can asynchronously become sealed.
+     *  
      *  @param name       name of the channel
      *  @return           the ivar with the channel registered under the specified name
      */
@@ -265,20 +305,19 @@ object IsoSystem {
 
     /** A default implementation of the channels register.
      */
-    class Default extends IsoSystem.Channels {
+    class Default(system: IsoSystem) extends IsoSystem.Channels(system) {
       private val channelMap = container.RMap[String, Channel[_]]
-      def update(name: String, c: Channel[_]) = channelMap.synchronized {
+      private[reactive] def update(name: String, c: Channel[_]) = channelMap.synchronized {
         if (!channelMap.contains(name)) channelMap(name) = c
         else sys.error(s"Name $name already contained in channels.")
       }
-      def apply[@spec(Int, Long, Double) T](name: String): Channel[T] = channelMap.synchronized {
-        channelMap(name).asInstanceOf[Channel[T]]
-      }
-      def remove(name: String): Unit = channelMap.synchronized {
+      private[reactive] def remove(name: String): Unit = channelMap.synchronized {
         channelMap.remove(name)
       }
-      def get[T](name: String): Option[Channel[T]] = channelMap.synchronized {
-        channelMap.get(name).asInstanceOf[Option[Channel[T]]]
+      private[reactive] def removeIsolate(isoName: String): Unit = channelMap.synchronized {
+        val prefix = isoName + "#"
+        val obsoleteNames = channelMap.keys.filter(_ startsWith prefix)
+        for (name <- obsoleteNames) channelMap.remove(name)
       }
       private def channelExtractor[T](reqId: Long): PartialFunction[InternalEvent, Channel[T]] = {
         case ChannelRetrieved(`reqId`, c: Channel[T]) => c

src/main/scala/scala/reactive/Proto.scala

@@ -17,7 +17,8 @@ final class Proto[+I <: Iso[_]] private[reactive] (
   val params: Seq[Any],
   val scheduler: String = null,
   val eventQueueFactory: EventQueue.Factory = null,
-  val multiplexer: Multiplexer = null
+  val multiplexer: Multiplexer = null,
+  val name: String = null
 ) {
 
   /** Instantiates and returns the isolate.
@@ -31,17 +32,22 @@ final class Proto[+I <: Iso[_]] private[reactive] (
    *  @param sname               name of the scheduler
    *  @return                    a new `Proto` object
    */
-  def withScheduler(sname: String): Proto[I] = new Proto(clazz, params, sname, eventQueueFactory, multiplexer)
+  def withScheduler(sname: String): Proto[I] = new Proto(clazz, params, sname, eventQueueFactory, multiplexer, name)
 
   /** Associates the specified event queue type and returns the new `Proto` object.
    *  
    *  @param f                   event queue factory, used to instantiate the event queue object
    *  @return                    a new `Proto` object
    */
-  def withEventQueue(f: EventQueue.Factory): Proto[I] = new Proto(clazz, params, scheduler, f, multiplexer)
+  def withEventQueue(f: EventQueue.Factory): Proto[I] = new Proto(clazz, params, scheduler, f, multiplexer, name)
 
+  /** Associates a multiplexer with the event queue type and returns the new `Proto` object.
+   */
+  def withMultiplexer(m: Multiplexer): Proto[I] = new Proto(clazz, params, scheduler, eventQueueFactory, m, name)
 
-  def withMultiplexer(m: Multiplexer): Proto[I] = new Proto(clazz, params, scheduler, eventQueueFactory, m)
+  /** Associates the name for the new isolate and returns the new `Proto` object.
+   */
+  def withName(nm: String): Proto[I] = new Proto(clazz, params, scheduler, eventQueueFactory, multiplexer, nm)
 
 }
 

src/main/scala/scala/reactive/container/RMap.scala

@@ -14,6 +14,10 @@ trait RMap[@spec(Int, Long, Double) K, V <: AnyRef] extends RContainer[(K, V)] {
 
   def entries: PairContainer[K, V]
 
+  def keys: RContainer[K]
+
+  def values: RContainer[V]
+
   def react: RMap.Lifted[K, V]
 
 }

src/main/scala/scala/reactive/isolate/Connector.scala

@@ -15,11 +15,13 @@ package isolate
  *  @tparam T            the type of the events in this connector
  *  @param frame         the isolate frame
  *  @param queue         the event queue
+ *  @param name          the name of the channel in this connector
  *  @param isDaemon      is the connector a daemon -- daemon channels are ignored when determining termination
  */
 class Connector[@spec(Int, Long, Double) T](
   private[reactive] val frame: IsoFrame,
   private[reactive] val queue: EventQueue[T],
+  val name: String,
   val isDaemon: Boolean
 ) {
   @volatile private[reactive] var dequeuer: Dequeuer[T] = _
@@ -73,6 +75,7 @@ object Connector {
     }
     def unreact() = {
       multiplexer.unreacted(connector)
+      connector.frame.isolateSystem.channels.removeIsolate(connector.name)
       connector.frame.apply()
     }
   }

src/main/scala/scala/reactive/isolate/DefaultIsoSystem.scala

@@ -34,18 +34,19 @@ extends IsoSystem {
     else name
   }
 
-  protected[reactive] def releaseName(name: String): Unit = monitor.synchronized {
+  protected[reactive] def releaseNames(name: String): Unit = monitor.synchronized {
+    channels.removeIsolate(name)
     isolates.remove(name)
   }
 
   protected[reactive] def newChannel[@spec(Int, Long, Double) Q](reactor: Reactor[Q]) = {
     new Channel.Synced(reactor, new util.Monitor)
   }
 
-  val channels = new IsoSystem.Channels.Default
+  val channels = new IsoSystem.Channels.Default(this)
 
-  def isolate[@spec(Int, Long, Double) T: Arrayable](proto: Proto[Iso[T]], name: String = null): Channel[T] = {
-    val isolate = createFrame(proto, name)
+  def isolate[@spec(Int, Long, Double) T: Arrayable](proto: Proto[Iso[T]]): Channel[T] = {
+    val isolate = createFrame(proto)
     val frame = isolate.frame
     monitor.synchronized {
       isolates(frame.name) = frame

src/main/scala/scala/reactive/isolate/IsoFrame.scala

@@ -96,7 +96,7 @@ final class IsoFrame(
       if (isolateState.compareAndSet(Running, Terminated)) {
         try isolate.systemEmitter += IsoTerminated
         finally try for (es <- isolate.eventSources) es.close()
-        finally isolateSystem.releaseName(name)
+        finally isolateSystem.releaseNames(name)
       } else checkTerminated()
     }
   }

src/main/scala/scala/reactive/isolate/Multiplexer.scala

@@ -117,8 +117,8 @@ object Multiplexer {
     }
 
     private def addConnector(connector: Connector[_]) {
-      descriptors += connector
       if (!connector.isDaemon) liveCount += 1
+      descriptors += connector
     }
 
     private def deleteCurrentConnector() {