typed actor refs

gistfile1.scala

package typedactorrefs

import akka.actor._
import akka.pattern.AskSupport
import akka.util.Timeout
import scala.concurrent.Future
import typedactorrefs.Server.ServerMessage
import scala.concurrent.duration._
import typedactorrefs.Server.GetModule
import typedactorrefs.Server.Module

object Main {
 def main(args: Array[String]) {
   val system = ActorSystem("System")
   system.actorOf(Props[Server])
   Thread.sleep(1000)
   system.shutdown()
 }
}

object Server {
  abstract sealed class ServerMessage
  case class GetModule(id: String) extends ServerMessage
  case class Module(module: ActorRef)
}

class Server extends Actor {

  import Server._

  private val modules = Map("one" -> context.actorOf(Props(new Module1(TypedActorRef[ServerMessage](self)))),
    "two" -> context.actorOf(Props(new Module2(TypedActorRef[ServerMessage](self)))))

  def receive: Receive = {
    case GetModule(id) => sender ! Module(modules(id))
  }
}

object Module1 {
  abstract sealed trait Module1Message
  case object Hi extends Module1Message
  case class Text(text: String) extends Module1Message
}

class Module1(val server: TypedActorRef[ServerMessage]) extends Actor with ModuleProvider {
  import Module1._
  import Module2._

  override def preStart() {
    import scala.concurrent.ExecutionContext.Implicits.global
     getModule[Module2Message]("two") onSuccess {
       case m =>
         // m ! "test" does not compile!
        m ! MoreText("Hello, World")
     }
  }

  def receive: Receive = {
    case Hi => println("Hi")
    case Text(text) => println(text)
  }
}

object Module2 {
  abstract sealed trait Module2Message
  case object Bye extends Module2Message
  case class MoreText(text: String) extends Module2Message
}

class Module2(val server: TypedActorRef[ServerMessage]) extends Actor {
  import Module2._

  def receive: Receive = {
    case Bye => println("Bye")
    case MoreText(text) => println(text)
  }
}

final case class TypedActorRef[T](actorRef: ActorRef) extends AnyVal with java.lang.Comparable[ActorRef] with Serializable {

  def path: ActorPath = actorRef.path

  def compareTo(other: ActorRef) = this.path compareTo other.path

  def tell(msg: T, sender: ActorRef) {
    actorRef.tell(msg, sender)
  }

  def forward(message: T)(implicit context: ActorContext) = actorRef.forward(message)(context)

  def isTerminated: Boolean = actorRef.isTerminated

  override def toString = "Actor[%s]".format(path)

  def !(message: T)(implicit sender: ActorRef = Actor.noSender) {
    actorRef.!(message)(sender)
  }
}

object TypedAskSupport extends AskSupport {
  implicit def ask[T](typedActorRef: TypedActorRef[T]): TypedAskableActorRef[T] = new TypedAskableActorRef[T](typedActorRef)

  private[typedactorrefs] final class TypedAskableActorRef[T](val typedActorRef: TypedActorRef[T]) {

    def ask(message: T)(implicit timeout: Timeout): Future[Any] = akka.pattern.ask(typedActorRef.actorRef, message)(timeout)

    def ?(message: T)(implicit timeout: Timeout): Future[Any] = akka.pattern.ask(typedActorRef.actorRef, message)(timeout)
  }
}

trait ModuleProvider {
  implicit val timeout = Timeout(5 seconds)

  protected def server: TypedActorRef[ServerMessage]

  protected def getModule[T](moduleID: String): Future[TypedActorRef[T]] = {
    import typedactorrefs.TypedAskSupport.ask
    import scala.concurrent.ExecutionContext.Implicits.global
    (server ? GetModule(moduleID)).mapTo[Module].map(m => TypedActorRef[T](m.module))
  }
}