ChatRoom.scala 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. package models
  2. import akka.actor._
  3. import scala.concurrent.duration._
  4. import scala.language.postfixOps
  5. import play.api._
  6. import play.api.libs.json._
  7. import play.api.libs.iteratee._
  8. import play.api.libs.concurrent._
  9. import akka.util.Timeout
  10. import akka.pattern.ask
  11. import play.api.Play.current
  12. import play.api.libs.concurrent.Execution.Implicits._
  13. object ChatRoom {
  14. implicit val timeout = Timeout(1 second)
  15. lazy val default = Akka.system.actorOf(Props[ChatRoom])
  16. def join(username:String):scala.concurrent.Future[(Iteratee[JsValue,_],Enumerator[JsValue])] = {
  17. (default ? Join(username)).map {
  18. case Connected(enumerator) =>
  19. // Create an Iteratee to consume the feed
  20. val iteratee = Iteratee.foreach[JsValue] { event =>
  21. default ! Talk(username, (event \ "text").as[String])
  22. }.map { _ =>
  23. default ! Quit(username)
  24. }
  25. (iteratee,enumerator)
  26. case CannotConnect(error) =>
  27. // Connection error
  28. // A finished Iteratee sending EOF
  29. val iteratee = Done[JsValue,Unit]((),Input.EOF)
  30. // Send an error and close the socket
  31. val enumerator = Enumerator[JsValue](JsObject(Seq("error" -> JsString(error)))).andThen(Enumerator.enumInput(Input.EOF))
  32. (iteratee,enumerator)
  33. }
  34. }
  35. }
  36. class ChatRoom extends Actor {
  37. val tabooGame = Akka.system.actorOf(Props(classOf[TabooGame], self))
  38. var members = Map.empty[String, Concurrent.Channel[JsValue]]
  39. val (chatEnumerator, chatChannel) = Concurrent.broadcast[JsValue]
  40. def receive = {
  41. case Join(username) => {
  42. if(members.contains(username)) {
  43. sender ! CannotConnect("This username is already used")
  44. } else {
  45. val (personalEnumerator, personalChannel) = Concurrent.broadcast[JsValue]
  46. members = members + (username -> personalChannel)
  47. sender ! Connected(chatEnumerator.interleave(personalEnumerator))
  48. tabooGame ! Join(username)
  49. }
  50. }
  51. case Talk(username, "/ping") => {
  52. self ! Tell(username, Json.obj(
  53. "kind" -> "pong"
  54. ))
  55. }
  56. case Talk(username, text) => {
  57. if(!text.startsWith("/")) {
  58. self ! Announce(Json.obj(
  59. "kind" -> "talk",
  60. "user" -> username,
  61. "message" -> text
  62. ))
  63. }
  64. tabooGame ! Talk(username, text)
  65. }
  66. case Announce(message) => {
  67. chatChannel.push(message)
  68. }
  69. case Tell(username, message) => {
  70. members(username).push(message)
  71. }
  72. case Quit(username) => {
  73. members = members - username
  74. tabooGame ! Quit(username)
  75. }
  76. }
  77. }
  78. case class Join(username: String)
  79. case class Quit(username: String)
  80. case class Talk(username: String, text: String)
  81. case class Announce(value: JsValue)
  82. case class Tell(username: String, value: JsValue)
  83. case class Connected(enumerator:Enumerator[JsValue])
  84. case class CannotConnect(msg: String)