example of Concurent.unicast in Play 2.1

unicast.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

object Movies extends Controller {
  import scala.concurrent.stm._
  val sells = Ref(Map("1" -> 0l, "2" -> 0l, "3" -> 0l))

  def buy(id: String) = Action {
    Actors.notifier ! Event("jto", "movies.buy", id)
    atomic { implicit txn =>
      val ms = sells()
      val updated: Map[String, Long] = ms + (id -> ms.get(id).map(_ + 1).getOrElse(1))
      sells()= updated

      val tops = updated.toSeq.sortBy(_._2).map(_._1)
      Actors.notifier ! Event("jto", "movies.tops", Json.toJson(tops).toString)
    }
    Ok
  }

  def vod = Action {
    Ok(views.html.vod("VOD"))
  }
}

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

object Application extends Controller {
  import Actors._

  val username = "jto"
  implicit val timeout = Timeout(1 second)

  def index(content: String) = Action {
    val r = (a: play.api.mvc.Call) => Ok(views.html.main("Hello Bouygues", a))
    content match {
      case "address" => r(routes.Application.address)
      case "vod" => r(routes.Movies.vod)
      case _ => NotFound
    }
  }

  def address = Action {
    Ok(views.html.index())
  }

  def saveAdress = Action {
    notifier ! Event(username, "adress.updating", """updating""")
    Promise.timeout(":)", 5 seconds).map{ _ =>
      notifier ! Event(username, "adress.updated", """done""")
    }
    Ok
  }

  def notif(username: String, kind: String, txt: String) = Action {
    notifier ! Event(username, kind, txt)
    Ok
  }

  def event = WebSocket.async[JsValue] { request  =>
    (notifier ? Join(username)).map {
      case Connected(enumerator) =>

        val iteratee = Iteratee.foreach[JsValue] { event =>
          play.Logger.info("received: " + event)
        }.mapDone { _ =>
          notifier ! Quit(username)
        }
        (iteratee, enumerator)
    }
  }
}

class Events extends Actor {
  var connected = Map.empty[String, Concurrent.Channel[JsValue]]

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

    case Quit(username) => {
      connected = connected - username
    }

    case Event(user, kind, text) => {
      for(channel <- connected.get(user)){
        val msg = JsObject(
          Seq(
            "kind" -> JsString(kind),
            "text" -> JsString(text)))
        channel.push(msg)
      }
    }
  }
}

case class Event(user: String, kind: String, text: String)
case class Connected(enumerator: Enumerator[JsValue])
case class Join(username: String)
case class Quit(username: String)