ChatRoom.scala 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  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. var chatRooms = Map.empty[String, ActorRef]
  16. def closeRoom(room: String) = {
  17. chatRooms -= room
  18. }
  19. def join(room: String, username:String):scala.concurrent.Future[(Iteratee[JsValue,_],Enumerator[JsValue])] = {
  20. if(!chatRooms.keySet(room)) {
  21. chatRooms += (room -> Akka.system.actorOf(Props(classOf[ChatRoom], room)))
  22. }
  23. val actor = chatRooms(room)
  24. (actor ? Join(username)).map {
  25. case Connected(enumerator) =>
  26. // Create an Iteratee to consume the feed
  27. val iteratee = Iteratee.foreach[JsValue] { event =>
  28. actor ! Talk(username, (event \ "text").as[String])
  29. }.map { _ =>
  30. actor ! Quit(username)
  31. }
  32. (iteratee,enumerator)
  33. case CannotConnect(error) =>
  34. // Connection error
  35. // A finished Iteratee sending EOF
  36. val iteratee = Done[JsValue,Unit]((),Input.EOF)
  37. // Send an error and close the socket
  38. val enumerator = Enumerator[JsValue](JsObject(Seq("error" -> JsString(error)))).andThen(Enumerator.enumInput(Input.EOF))
  39. (iteratee,enumerator)
  40. }
  41. }
  42. }
  43. class ChatRoom(name: String) extends Actor {
  44. val tabooGame = Akka.system.actorOf(Props(classOf[TabooGame], self))
  45. var members = Map.empty[String, Concurrent.Channel[JsValue]]
  46. val (chatEnumerator, chatChannel) = Concurrent.broadcast[JsValue]
  47. def receive = {
  48. case Join(username) => {
  49. if(members.contains(username)) {
  50. sender ! CannotConnect("This username is already used")
  51. } else {
  52. val (personalEnumerator, personalChannel) = Concurrent.broadcast[JsValue]
  53. members = members + (username -> personalChannel)
  54. sender ! Connected(chatEnumerator.interleave(personalEnumerator))
  55. tabooGame ! Join(username)
  56. }
  57. }
  58. case Talk(username, "/ping") => {
  59. self ! Tell(username, Json.obj(
  60. "kind" -> "pong"
  61. ))
  62. }
  63. case Talk(username, text) => {
  64. if(!text.startsWith("/")) {
  65. self ! Announce(Json.obj(
  66. "kind" -> "talk",
  67. "user" -> username,
  68. "message" -> text
  69. ))
  70. }
  71. tabooGame ! Talk(username, text)
  72. }
  73. case Announce(message) => {
  74. chatChannel.push(message)
  75. }
  76. case Tell(username, message) => {
  77. members(username).push(message)
  78. }
  79. case Quit(username) => {
  80. members = members - username
  81. if(members.isEmpty) {
  82. ChatRoom.closeRoom(name)
  83. context.stop(tabooGame)
  84. context.stop(self)
  85. }
  86. else {
  87. tabooGame ! Quit(username)
  88. }
  89. }
  90. }
  91. }
  92. case class Join(username: String)
  93. case class Quit(username: String)
  94. case class Talk(username: String, text: String)
  95. case class Announce(value: JsValue)
  96. case class Tell(username: String, value: JsValue)
  97. case class Connected(enumerator:Enumerator[JsValue])
  98. case class CannotConnect(msg: String)