Actor migration instantiation tests and corresponding minor fixes.

Author: Vojin Jovanovic

src/actors/scala/actors/Actor.scala

@@ -313,11 +313,7 @@ object Actor extends Combinators {
         def respond(k: B => Unit) = fun(caseBlock andThen k)
       }
 
-  private[actors] trait Body[a] {
-    def andThen[b](other: => b): Unit
-  }
-
-  implicit def mkBody[a](body: => a) = new Body[a] {
+  implicit def mkBody[a](body: => a) = new InternalActor.Body[a] {
     def andThen[b](other: => b): Unit = rawSelf.seq(body, other)
   }    
 

src/actors/scala/actors/Combinators.scala

@@ -16,7 +16,7 @@ private[actors] trait Combinators {
    * Enables the composition of suspendable closures using `andThen`,
    * `loop`, `loopWhile`, etc.
    */
-  implicit def mkBody[a](body: => a): Actor.Body[a]
+  implicit def mkBody[a](body: => a): InternalActor.Body[a]
 
   /**
    * Repeatedly executes `body`.

src/actors/scala/actors/InternalActor.scala

@@ -10,6 +10,12 @@ import java.util.TimerTask
 import scala.util.continuations.suspendable
 import scala.util.control.ControlThrowable
 
+private[actors] object InternalActor {
+  private[actors] trait Body[a] {
+    def andThen[b](other: => b): Unit
+  }
+}
+
 private[actors] trait InternalActor extends AbstractActor with InternalReplyReactor with ActorCanReply with InputChannel[Any] with Serializable {
 
   /* The following two fields are only used when the actor
@@ -432,7 +438,8 @@ private[actors] trait InternalActor extends AbstractActor with InternalReplyReac
 
   private[actors] def internalPostStop() = {}
 
-  private[actors] def stop(reason: AnyRef): Unit = {
+  
+  private[actors] def stop(reason: AnyRef): Unit = {    
     synchronized {
       shouldExit = true
       exitReason = reason
@@ -444,7 +451,8 @@ private[actors] trait InternalActor extends AbstractActor with InternalReplyReac
       else if (waitingFor ne Reactor.waitingForNone) {
         waitingFor = Reactor.waitingForNone
         // it doesn't matter what partial function we are passing here
-        scheduleActor(waitingFor, null)
+        val task = new ActorTask(this, null, waitingFor, null)
+        scheduler execute task                          
         /* Here we should not throw a SuspendActorControl,
            since the current method is called from an actor that
            is in the process of exiting.

src/actors/scala/actors/Reactor.scala

@@ -259,7 +259,7 @@ trait Reactor[Msg >: Null] extends OutputChannel[Msg] with Combinators {
       _state
   }
 
-  implicit def mkBody[A](body: => A) = new Actor.Body[A] {
+  implicit def mkBody[A](body: => A) = new InternalActor.Body[A] {
     def andThen[B](other: => B): Unit = Reactor.this.seq(body, other)
   }
 

src/actors/scala/actors/RichActor.scala

@@ -3,8 +3,14 @@ package scala.actors
 import scala.util.continuations._
 import scala.collection._
 
+object RichActor extends Combinators {
+  implicit def mkBody[A](body: => A) = new InternalActor.Body[A] {
+    def andThen[B](other: => B): Unit = Actor.rawSelf.seq(body, other)
+  }
+}
+
 trait RichActor extends InternalActor {
-  type Receive = PartialFunction[Any, Unit @suspendable]
+  type Receive = PartialFunction[Any, Unit]
 
   // checks if RichActor is created within the actorOf block
   creationCheck;
@@ -173,10 +179,10 @@ trait RichActor extends InternalActor {
     // creation check (see ActorRef)
     val context = ActorSystem.contextStack.get
     if (context.isEmpty)
-      throw new Exception("must use actorOf to create actor")
+      throw new RuntimeException("In order to create RichActor one must use actorOf.")
     else {
       if (!context.head)
-        throw new Exception("must use actorOf to create actor")
+        throw new RuntimeException("Only one actor can be created per actorOf call.")
       else
         ActorSystem.contextStack.set(context.push(false))
     }
@@ -191,23 +197,22 @@ trait RichActor extends InternalActor {
    * Method that models the behavior of Akka actors.  
    */
   private[actors] def internalAct() {
-    reset {
+    
+    behaviorStack = behaviorStack.push(new PartialFunction[Any, Unit] {
+      def isDefinedAt(x: Any) =
+        handle.isDefinedAt(x)
+      def apply(x: Any) = handle(x)              
+    } orElse {
+      case m => unhandled(m)
+    })
 
-      behaviorStack = behaviorStack.push(new PartialFunction[Any, Unit] {
-        def isDefinedAt(x: Any) =
-          handle.isDefinedAt(x)
-        def apply(x: Any) =
-          reset { handle(x) }
-      } orElse {
-        case m => unhandled(m)
-      })
-
-      while (true)
+    reset {
+      while(true) {      
         if (receiveTimeout.isDefined)
-          reactWithin(receiveTimeout.get)(behaviorStack.top) // orElse 
+          reactWithin(receiveTimeout.get)(behaviorStack.top) 
         else
           react(behaviorStack.top)
-
+      }
     }
   }
 

src/actors/scala/actors/akka/ActorRef.scala

@@ -21,7 +21,6 @@ trait ActorRef {
 
   def start(): ActorRef
 
-  // TODO (VJ) test with linked
   /**
    * Shuts down the actor its dispatcher and message queue.
    */
@@ -105,8 +104,9 @@ private[actors] class ReplyActorRef(override val actor: InternalReplyReactor) ex
 
 }
 
-private[actors] final class RichActorRef(override val actor: RichActor) extends ReplyActorRef(actor) {
+private[actors] final class InternalActorRef(override val actor: InternalActor) extends ReplyActorRef(actor) {
 
+  // TODO (VJ) this does not work
   override def stop(): Unit = actor.stop('normal)
 
 }

src/actors/scala/actors/akka/ActorSystem.scala

@@ -25,19 +25,16 @@ object ActorSystem {
         contextStack.set(if (!stackAfter.head) stackAfter.pop.pop else stackAfter.pop)
     }
   }
-
-  // actorOf(new MyActor())
-  def actorOf(factory: ⇒ RichActor): ActorRef = withCleanContext {
-    val r = new RichActorRef(factory)
+  
+  def actorOf(factory: ⇒ InternalActor): ActorRef = withCleanContext {
+    val r = new InternalActorRef(factory)
     r
   }  
 
-  def actorOf[T <: RichActor](implicit m: Manifest[T]): ActorRef = withCleanContext {
+  def actorOf[T <: InternalActor](implicit m: Manifest[T]): ActorRef = withCleanContext {
     val clazz = m.erasure.asInstanceOf[Class[_ <: RichActor]]
-    val r = new RichActorRef(clazz.newInstance())    
+    val r = new InternalActorRef(clazz.newInstance())    
     r
   }
 
-    
-  // TODO (VJ) def for plain actors 
 }

test/files/jvm/actmig-instantiation.check

@@ -0,0 +1,8 @@
+OK error: java.lang.RuntimeException: In order to create RichActor one must use actorOf.
+OK error: java.lang.RuntimeException: Only one actor can be created per actorOf call.
+0
+100
+200
+300
+400
+500

test/files/jvm/actmig-instantiation.scala

@@ -0,0 +1,93 @@
+import scala.actors.ActorSystem._
+import scala.actors.Actor._
+import scala.actors.{Actor, RichActor, ActorRef}
+import scala.util.continuations._
+import java.util.concurrent.{TimeUnit, CountDownLatch}
+import scala.collection.mutable.ArrayBuffer
+
+class TestRichActor extends RichActor {
+  
+  def handle = { case v: Int => Test.append(v); Test.latch.countDown() }
+
+}
+
+object Test {
+ val NUMBER_OF_TESTS = 5 
+
+ // used for sorting non-deterministic output
+  val buff = ArrayBuffer[Int](0)
+  val latch = new CountDownLatch(NUMBER_OF_TESTS)
+  val toStop = ArrayBuffer[ActorRef]()
+
+  def append(v: Int) = synchronized {
+    buff += v
+  }
+
+  def main(args: Array[String]) = {
+    // plain scala actor
+    val a1 = actor {
+      reset {react { case v:Int => Test.append(v); Test.latch.countDown() }}
+    }
+    a1 ! 100
+ 
+    // simple instantiation
+    val a2 = actorOf(new TestRichActor) 
+    a2.start()
+    a2 ! 200
+    
+    toStop += a2
+    // actor of with scala actor
+    val a3 = actorOf(actor{
+      reset {react { case v:Int => Test.append(v); Test.latch.countDown() }}
+    })        
+    a3 ! 300 
+     
+    // using the manifest    
+    val a4 = actorOf[TestRichActor].start()
+    a4 ! 400
+
+    toStop += a4
+    // deterministic part of a test 
+    
+    // creation without actorOf 
+    try {
+     val a3 = new TestRichActor
+     a3 ! -1
+    } catch {
+      case e => println("OK error: " + e)
+    }
+ 
+    // actorOf double creation
+    try {
+     val a3 = actorOf {
+       new TestRichActor
+       new TestRichActor
+     }
+     a3 ! -1
+    } catch {
+      case e => println("OK error: " + e)
+    }
+    
+    // actorOf nesting
+    try {
+     val a5 = actorOf {
+       val a6 = actorOf[TestRichActor]       
+       new TestRichActor
+     }
+     a5.start()
+     a5 ! 500
+     toStop += a5
+    } catch {
+      case e => println("Should not throw an exception: " + e)
+    }
+    
+    // output
+    latch.await(10, TimeUnit.MILLISECONDS)
+    if (latch.getCount() > 0) {
+      println("Error: Tasks have not finished!!!")
+    }
+
+    buff.sorted.foreach(println)
+    toStop.foreach(_.stop())
+  }
+}