Example of Exception Handling in Actors.

App.scala

package pt.uc.dei

import scala.collection.immutable.List
import scala.actors.Actor
import scala.actors.Actor._

case class MessageNotSent(val msg:String) extends Exception {
  override def getMessage = msg
}

object App {
  def main(args : Array[String]) {
      messageController.start
  }
}

object messageController extends Actor with ExceptionModel {   
  def act() {
    println("[Controller] Starting to send messages..")
    
    // initialize the system
    ExceptionController.start

    def handle(e:Exception) = {
      println("[Controller] Exception caught: " + e.getMessage)
    }
    
    (1 to 15).foreach { m =>
      val ms = new MessageSender
      ms.start
      ms ! m
      ms ! Stop
      
      checkFor[MessageNotSent](handle)
    }
    
    // Stop the system
    ExceptionController ! Stop
  }
}

class MessageSender extends Actor with ExceptionModel{   
  def act() {
    loop {
      react {
        case Stop => {
          exit()
        }
        case i:Int => {
          println("Sending Message " + i)
          if (i == 5) {
            // Message 5 fails assynchronously
            raise(new MessageNotSent(i.toString))
            println("Message " + i + " failed, but I went on.")
            
          } else if (i == 10) {
            // Message 10 fails synchronously
            raiseAndWait(new MessageNotSent(i.toString))
            println("Message " + i + " was taken care by the sender.")
          }
        }
        case _ => ()
      }
    }
  }
}

Model.scala

package pt.uc.dei

import scala.collection.mutable.{Queue,HashMap}
import scala.actors.Actor
import scala.actors.OutputChannel
import scala.actors.Actor._

case object Stop
case class Ack
case class Query[E](val f:Exception => Unit) {
  def test(e:Exception):Boolean = e.isInstanceOf[E]
}
case class SynchronousException(e:Exception)


trait ExceptionModel extends Actor {
  def raise(e:Exception) = {
    ExceptionController ! e
  }
  
  def raiseAndWait(e:Exception) = {
    ExceptionController !? new SynchronousException(e)
  }
  
  def checkFor[E <: Exception](f:Exception => Unit) = {
    ExceptionController !? new Query[E](f)
  }
}


object ExceptionController extends Actor {
  val exceptions = new Queue[Exception]();
  val replies = new HashMap[Exception,OutputChannel[Any]]();
  
  def act() {
    loop {
      react {
        case e:Exception => {
          this << e
        }
        case SynchronousException(e) => {
          exceptions.synchronized {
            replies.put(e,sender)
          }
          this << e
        }
        case q:Query[_] => {
          checkFor(q.test, q.f)
          sender ! Ack
        }
        case Stop => exit()
        case _ => println("erro :(")
      }
    }
  }
  
  def << (e:Exception) = {
    exceptions += e
  }
  
  def checkFor(test:Exception => Boolean, f:Exception => Unit) = {
    exceptions.synchronized {
      val found = exceptions.dequeueAll(test)
      found.foreach{
        e => if (replies.contains(e)) {
          replies.get(e) match {
            case Some(snd) => snd ! Ack
            case None => ()
          }
        } 
      }
      found.foreach(f)
    }
  }
}