simple #json messaging using #websocket and #akka #actors.

actors.scala

package controllers

import akka.actor._
import akka.pattern.ask
import akka.util.Timeout

import scala.concurrent.duration._
import scala.concurrent.Future

import play.api._
import play.api.mvc._
import play.api.libs.json._
import play.api.libs.iteratee._
import play.api.libs.concurrent._
import play.api.libs.concurrent.Execution.Implicits._
import play.api.Play.current

import models._

object Actors {
  lazy val notifier = Akka.system.actorOf(Props[Events])
}

object Implicits {
  implicit val now = new java.util.Date
  implicit val dateFormat = new java.text.SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss")
  implicit val timeout = Timeout(1 second)
}

class Events extends Actor {
  import Implicits._

  var connected = Map.empty[String, Seq[(Long, Concurrent.Channel[JsValue])]]

  def receive = {
    case Join(username, requestid) => {
      val e = Concurrent.unicast[JsValue]{ c =>
        play.Logger.info("Start")
        val channels = connected.get(username).toSeq.flatten :+ (requestid, c)
        connected = connected + (username -> channels)
        play.Logger.info("Connected: " + connected)
      }
      sender ! Connected(e)
    }

    case Quit(username, requestid) => {
      play.Logger.info("Quit: " + username)
      val channels = connected.get(username).toSeq.flatten.filter(_._1 != requestid)
      if(channels.isEmpty)
        connected = connected - username
      else
        connected = connected + (username -> channels)
    }

    case Event(user, kind, text) => {

      val channels: Seq[Concurrent.Channel[JsValue]] = user match {
        case ALL => connected.values.toSeq.flatten.map(_._2)
        case User(name) => connected.get(name).toSeq.flatten.map(_._2)
      }

      for(channel <- channels){
        val msg = JsObject(
          Seq(
            "kind" -> JsString(kind),
            "text" -> JsString(text)))
        channel.push(msg)
      }
    }
  }
}

case class Event(client: Client, kind: String, text: String)
case class Connected(enumerator: Enumerator[JsValue])
case class Join(username: String, requestid: Long)
case class Quit(username: String, requestid: Long)

controller.scala

package controllers

import akka.actor._
import akka.pattern.ask
import akka.util.Timeout

import scala.concurrent.duration._
import scala.concurrent.Future

import play.api._
import play.api.mvc._
import play.api.libs.json._
import play.api.libs.iteratee._
import play.api.libs.concurrent._
import play.api.libs.concurrent.Execution.Implicits._
import play.api.Play.current

import models._

object Com extends Controller {
  import Actors._
  import Implicits._

  def event = WebSocket.async[JsValue] { request  =>
    val user = request.session.get("username")
    user.map { u =>
      (notifier ? Join(u, request.id)).map {
        case Connected(enumerator) =>
          val iteratee = Iteratee.foreach[JsValue] { event =>
            play.Logger.info("received: " + event)
          }.mapDone { _ =>
            notifier ! Quit(u, request.id)
          }
          (iteratee, enumerator)
        }
    }.getOrElse( Future(Done(()) -> Enumerator(JsNull)))
  }
}