Minimalist Scala Actors (typed version)

minscalaactors.scala

/*
Copyright 2012 Viktor Klang
 
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at
 
       http://www.apache.org/licenses/LICENSE-2.0
 
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

object Actor {
  import java.util.concurrent.{ConcurrentLinkedQueue, Executor}
  import java.util.concurrent.atomic.AtomicInteger
  type Behavior[T] = T => Effect[T]
  def behavior[T](f: Actor.Behavior[T]): Actor.Behavior[T] = f
  sealed trait Effect[T] extends (Behavior[T] => Behavior[T])
  final def Stay[T] = new Effect[T] { def apply(old: Behavior[T]): Behavior[T] = old }
  case class Become[T](like: Behavior[T]) extends Effect[T] { def apply(old: Behavior[T]): Behavior[T] = like }
  // Stay Dead plz
  final def Die[T] = Become[T](msg => { println("Dropping msg [" + msg + "] due to severe case of death."); Stay})
  // The notion of an Address to where you can post messages to
  trait Address[-T] { def !(msg: T): Unit }
  private abstract class AtomicRunnableAddress[-T] extends Address[T] with Runnable { val on = new AtomicInteger(0) }
  // Seeded by the self-reference that yields the initial behavior
  def apply[T](initial: Address[T] => Behavior[T])(implicit e: Executor): Address[T] =
    // Memory visibility of "behavior" is guarded by "on" using volatile piggybacking
    new AtomicRunnableAddress[T] {
      // Our awesome little mailbox, free of blocking and evil
      private final val mbox = new ConcurrentLinkedQueue[T]
      // Rebindable top of the mailbox, bootstrapped to identity
      private var behavior: Behavior[T] = {
        case self: Address[_] => Become(initial(self.asInstanceOf[Address[T]]))
      }
      // As an optimization, we peek at our threads local copy
      // of our behavior to see if we should bail out early
      final override def !(msg: T): Unit = behavior match {
        // Efficiently bail out if we're _known_ to be dead
        //case dead @ Die.`like` => dead(msg.asInstanceOf[Nothing])
        // Enqueue the message onto the mailbox and try to schedule for execution
        case _ => mbox.offer(msg); async()
      }
      final def run(): Unit = try { if (on.get == 1) behavior = behavior(mbox.poll())(behavior) } finally { on.set(0); async() } // Switch ourselves off, and then see if we should be rescheduled for execution
      final def async(): Unit = if(!mbox.isEmpty && on.compareAndSet(0, 1))         // If there's something to process, and we're not already scheduled
        // Schedule to run on the Executor and back out on failure
        try e.execute(this) catch { case t: Throwable => on.set(0); throw t } // 
    } match {
      // Make the actor self aware by seeding its address to the initial
      case a: Address[T] => a ! a.asInstanceOf[T]; a
    }
}

import Actor._

implicit val e: java.util.concurrent.Executor = java.util.concurrent.Executors.newCachedThreadPool

type Natural = Actor.Address[Int]

val alice = Actor[(Natural, String)] { self =>
  behavior {
    case (sender, name) =>
      println(s"Alice: Oh! $name! I should die")
      sender ! 1
      Die
  }
}

val bob = Actor[Int] { self =>
  behavior {
    case 0 =>
      println(s"Bob: Hello, Alice!")
      alice ! (self, "Bob")
      Stay
    case 1 =>
      println(s"Bob: I should die too!")
      Die
  }
}

bob ! 0