Fixed bugs in orElse, more general typing.

2 files changed, 42 insertions, 15 deletions

diff --git a/src/actors/scala/actors/Actor.scala b/src/actors/scala/actors/Actor.scala
index b44326fe53..370b861bab 100644
--- a/src/actors/scala/actors/Actor.scala
+++ b/src/actors/scala/actors/Actor.scala
@@ -68,6 +68,8 @@ object Actor {
 
   def ? : Any = self.in.?
 
+  def poll: Option[Any] = self.in.poll
+
   /**
    * Receives a message from the mailbox of
    * <code>self</code>. Blocks if no message matching any of the
@@ -168,17 +170,17 @@ object Actor {
    */
   def reply(): Unit = reply(())
 
-  private[actors] trait Body[T] {
-    def orElse(other: => T): T
-    def andThen(other: => T): T
+  private[actors] trait Body[a] {
+    def orElse[b >: a](other: => b): b
+    def andThen[b >: a](other: => b): b
   }
 
-  implicit def mkBody(body: => Unit) = new Body[Unit] {
-    def orElse(other: => Unit): Unit = choose(body, other)
-    def andThen(other: => Unit): Unit = seq(body, other)
+  implicit def mkBody[a](body: => a) = new Body[a] {
+    def orElse[b >: a](other: => b): b = choose(body, other)
+    def andThen[b >: a](other: => b): b = seq(body, other)
   }
 
-  private[actors] def choose(alt1: => Unit, alt2: => Unit): Unit = {
+  private[actors] def choose[a, b >: a](alt1: => a, alt2: => b): b = {
     val s = self
     // save former custom suspendActor function
     // (e.g. from further orElse)
@@ -187,10 +189,22 @@ object Actor {
 
     // have to get out of the point of suspend in alt1's
     // receive
-    s.suspendActor = () => { throw new SuspendActorException }
-    s.detachActor = f => { throw new SuspendActorException }
+    s.suspendActor = () => {
+      s.in.isSuspended = false
+      s.in.waitingFor = s.in.waitingForNone
+      throw new SuspendActorException
+    }
+    s.detachActor = f => {
+      s.in.waitingFor = s.in.waitingForNone
+      throw new SuspendActorException
+    }
 
-    try { alt1 }
+    try {
+      val res = alt1
+      s.suspendActor = suspendNext
+      s.detachActor = detachNext
+      res
+    }
     catch {
       case d: SuspendActorException =>
         s.suspendActor = suspendNext
@@ -218,7 +232,7 @@ object Actor {
    * @param first ...
    * @param next  ...
    */
-  def seq(first: => Unit, next: => Unit): Unit = {
+  def seq[a, b >: a](first: => a, next: => b): b = {
     val s = self
     s.kill = () => { next }
     first
diff --git a/src/actors/scala/actors/Channel.scala b/src/actors/scala/actors/Channel.scala
index c064d96c73..74a5b88916 100644
--- a/src/actors/scala/actors/Channel.scala
+++ b/src/actors/scala/actors/Channel.scala
@@ -10,6 +10,8 @@
 
 package scala.actors
 
+import Actor._
+
 case object TIMEOUT
 
 class SuspendActorException extends Throwable {
@@ -41,9 +43,9 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
 
   private var received: Msg = _
 
-  private val waitingForNone = (m: Msg) => false
-  private var waitingFor: Msg => boolean = waitingForNone
-  private var waitingForSender: Actor = null
+  private[actors] val waitingForNone = (m: Msg) => false
+  private[actors] var waitingFor: Msg => boolean = waitingForNone
+  private[actors] var waitingForSender: Actor = null
 
   //private val messageQueue = new MessageQueue[Msg]
   private val mailbox = new scala.collection.mutable.Queue[Pair[Msg, Actor]]
@@ -62,6 +64,8 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
         TimerThread.trashRequest(receiver)
       }
 
+      // TODO: transmit call to protocol
+
       if (isSuspended)
         receiver.resumeActor()
       else
@@ -79,6 +83,12 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
 
   def ? : Msg = receive { case any => any }
 
+  def poll = {
+    Some(?)
+  } orElse {
+    None.asInstanceOf[Option[Msg]]
+  }
+
   /**
    * Sends <code>msg</code> to this <code>Channel</code> and
    * awaits reply.
@@ -97,7 +107,7 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
    */
   def forward(msg: Msg): unit = send(msg, receiver.sender)
 
-  private var isSuspended = false
+  private[actors] var isSuspended = false
 
   /**
    * Receives a message from this <code>Channel</code>.
@@ -126,6 +136,7 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
       }) match {
         case Some(Pair(msg, sender)) => {
           received = msg
+	  // TODO: call to delivery protocol action
           receiver.pushSender(sender)
         }
         case None => {
@@ -142,6 +153,7 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
     }
     receiver.resetActor()
     val result = f(received)
+    // TODO: call to complete protocol action
     receiver.popSender()
     result
   }
@@ -288,6 +300,7 @@ class Channel[Msg] extends InputChannel[Msg] with OutputChannel[Msg] {
           received = msg
           receiver.pushSender(sender)
           waitingFor = waitingForNone
+          // TODO: call to delivery protocol action
           receiver.scheduleActor(f, received)
         }
         case None => {