class RoomWriter(roomLogId: String) extends PersistentActor {

  import RoomWriter._

  override def persistenceId = roomLogId

  val listeners = mutable.Set.empty[ActorRef]

  def receiveRecover = Actor.emptyBehavior

  def receiveCommand = {
    case msg: Message =>
      persistAsync(msg) { _ =>
        listeners foreach (_ ! Update)
      }

    case Listen(ref) => listeners add context.watch(ref)

    case Terminated(ref) => listeners remove ref
  }
}

object RoomWriter {

  case class Listen(ref: ActorRef)

  case class Message(author: String, content: String, time: Long)

  case object Update

  def props(roomId: String) = Props(new RoomWriter(roomId))
}

class RoomReader(roomLogId: String, roomWriter: ActorRef, userConnection: ActorRef) extends PersistentView {
  import RoomWriter._

  roomWriter ! Listen(self)

  override def persistenceId = roomLogId

  override def viewId = roomLogId + "-view"

  def receive = {
    case msg @ Message(_, _,sendingTime) if currentTime - sendingTime < tenMinutes =>
      userConnection ! msg
    case msg: Message =>
    case Update => self ! akka.persistence.Update()
  }
}

object RoomReader {

  def currentTime = System.currentTimeMillis()

  val tenMinutes = Duration(10, MINUTES).toMillis

  def props(roomLogId: String, roomWriter: ActorRef, userConnection: ActorRef) = Props(
    new RoomReader(roomLogId, roomWriter, userConnection)
  )
}

class UserConnection(receptionist: ActorRef, roomWriter: ActorRef, out: ActorRef, roomLogId: String) extends Actor {
  import actors.UserConnection._

  def receive = waitingForUsername
  
  def waitingForUsername: Receive = {
    case WebSocketInMsg(RegisterMeWithName, username) => receptionist ! UsernameRequest(username)
    case Ack(username) =>
      context become readyToChat(username)
      context actorOf RoomReader.props(roomLogId, roomWriter, self)
      out ! WebSocketOutMsg(currentTime, "system", "welcome")
    case NAck => out ! WebSocketOutMsg(currentTime, "system", "taken")
  }

  def readyToChat(username: String): Receive = {
    case WebSocketInMsg(SendMessage, message) => roomWriter ! Message(username, message, currentMillis)
    case Message(author, content, time) => out ! WebSocketOutMsg(formatTime(time), author, content)
  }
}

object UserConnection {
  def props(receptionist: ActorRef, roomWriter: ActorRef, out: ActorRef, roomLogId: String) = Props(
    new UserConnection(receptionist, roomWriter, out, roomLogId)
  )

  case class WebSocketInMsg(messageType: Int, messageText: String)
  case class WebSocketOutMsg(time: String, from: String, messageText: String)

  case class UsernameRequest(name: String)
  case class Ack(username: String)
  case object NAck

  val RegisterMeWithName = 0
  val SendMessage = 1

  val formatter = DateTimeFormat.forPattern("HH:mm:ss").withLocale(Locale.US)
  def currentTime = DateTime.now().toString(formatter)
  def currentMillis = System.currentTimeMillis()
  def formatTime(timeStamp: Long) = new DateTime(timeStamp).toString(formatter)
}

class Receptionist extends Actor {

  var takenNames = mutable.Map("system" -> self)

  def receive = {
    case UsernameRequest(username) =>
      if (takenNames contains username) {
        sender() ! NAck
      } else {
        takenNames += (username -> context.watch(sender()))
        sender() ! Ack(username)
      }
    case Terminated(ref) => takenNames collectFirst {
      case (name, actor) if actor == ref => name
    } foreach takenNames.remove
  }
}

object Receptionist {
  def props() = Props[Receptionist]
}

object Application extends Controller {

  val logId = "akka-is-awesome"
  val roomWriter = Akka.system.actorOf(RoomWriter.props(logId), "writer")
  val receptionist = Akka.system.actorOf(Receptionist.props(), "receptionist")

  def index = Action { implicit request =>
    Ok(views.html.chat())
  }

  implicit val InMsgFormat = Json.format[WebSocketInMsg]
  implicit val InMsgFrameFormatter = FrameFormatter.jsonFrame[WebSocketInMsg]
  
  implicit val OutMsgFormat = Json.format[WebSocketOutMsg]
  implicit val OutMsgFrameFormatter = FrameFormatter.jsonFrame[WebSocketOutMsg]

  def socket = WebSocket.acceptWithActor[WebSocketInMsg, WebSocketOutMsg] { request => out =>
    UserConnection.props(receptionist, roomWriter, out, logId)
  }
}

angular.module('chatApp', [])
    .controller('ChatCtrl', ['$scope', function($scope) {
 
        var wsUri = "ws://"+window.location.host+"/ws";
        var websocket = new WebSocket(wsUri);
 
 
        $scope.name = "";
        $scope.messages = [];
        $scope.registered = false;
        $scope.taken = false;
        $scope.sendMessage = function () {
            websocket.send(angular.toJson({
                "messageType": 1,
                "messageText":$scope.messageText
            }));
            $scope.messageText = "";
        };
        $scope.sendName = function () {
            if (!$scope.registered) {
                websocket.send(angular.toJson({
                    "messageType": 0,
                    "messageText": $scope.name
                }));
            }
        };
 
        websocket.onmessage = function (e) {
            var msg = angular.fromJson(e.data);
            console.log(e.data);
            if (!$scope.registered) {
                switch (msg.from) {
                    case "system":
                        handleSystemMsg(msg.messageText);
                        break;
                }
            } else {
                $scope.messages.push(msg);
                $scope.$apply();
                var chatWindow = $("#chat-window");
                chatWindow.scrollTop(chatWindow[0].scrollHeight);
            }
        };
 
        function handleSystemMsg(msg) {
            switch (msg) {
                case "welcome":
                    $scope.registered = true;
                    break;
                case "taken":
                    $scope.taken = true;
                    break;
            }
        }
    }]);

<!DOCTYPE html>
<html ng-app="chatApp">
    <head>
        <meta charset="utf-8">
        <meta http-equiv="X-UA-Compatible" content="IE=edge">
        <meta name="viewport" content="width=device-width, initial-scale=1">
        <meta name="description" content="">
        <meta name="author" content="">
 
        <title>Akka WebSocket Chat</title>
            <!-- Bootstrap core CSS -->
        <link href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.1/css/bootstrap.min.css" rel="stylesheet">
        <script src="https://ajax.googleapis.com/ajax/libs/angularjs/1.3.5/angular.min.js"></script>
            <!-- Custom styles for this template -->
        <link href="@routes.Assets.at("stylesheets/main.css")" rel="stylesheet">
        <script src="@routes.Assets.at("javascripts/chatApp.js")"></script>
    </head>
 
    <body>
        <div ng-controller="ChatCtrl">
 
            <nav class="navbar navbar-inverse navbar-fixed-top" role="navigation">
                <div class="container">
                    <div class="navbar-header">
                        <a class="navbar-brand" href="#">Reactive Messenger</a>
                    </div>
                    <form class="navbar-form navbar-left" ng-submit="sendName()" ng-show="!registered">
                        <div class="form-group">
                            <input type="text" class="form-control" ng-model="name" placeholder="Username" required>
                        </div>
                        <button type="submit" class="btn btn-default">Set name</button>
                    </form>
                </div>
            </nav>
 
            <div class="container" >
                <div class="chat col-lg-6">
                    <div id="chat-window">
                        <ul class="list-group">
                            <li class="list-group-item" ng-repeat="message in messages">
                                <span class="label label-info">{{message.time}}</span>
                                <span class="label label-default">{{message.from}}</span> {{message.messageText}}
                            </li>
                        </ul>
                    </div>
                    <form ng-submit="sendMessage()">
                        <div>
                            <div class="input-group">
                                <input type="text" ng-model="messageText" class="form-control" required>
                                <span class="input-group-btn">
                                    <button class="btn btn-default" type="submit">
                                         Send<span class="glyphicon glyphicon-send" aria-hidden="true"></span>
                                    </button>
                                </span>
                            </div> <!-- /input-group -->
                        </div> <!-- /.col-lg-6 -->
                    </form>
                </div>
 
            </div> <!-- /.container -->
        </div>
 
 
        <!-- Bootstrap core JavaScript
    ================================================== -->
        <!-- Placed at the end of the document so the pages load faster -->
        <script src="https://ajax.googleapis.com/ajax/libs/jquery/1.11.1/jquery.min.js"></script>
        <script src="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.1/js/bootstrap.min.js"></script>
    </body>
</html>

akka.persistence.journal.plugin = "cassandra-journal"

cassandra-journal.contact-points = ["ip1,ip2,ip3"]

akka.persistence.snapshot-store.plugin = "cassandra-snapshot-store"

cassandra-snapshot-store.contact-points = ["ip1,ip2,ip3"]

option java_package = "actors.messages";
option optimize_for = SPEED;
message ChatMessage {
  optional string author = 1;
  optional string content = 2;
  optional int64 timestamp = 3;
}

class ChatMessageSerializer extends Serializer {

  def identifier: Int = 193823
 def includeManifest: Boolean = false

 def toBinary(obj: AnyRef): Array[Byte] = obj match {
    case ChatMessage(author, content, timestamp) =>
      ProtoChatMessage.newBuilder()
      .setAuthor(author)
      .setContent(content)
      .setTimestamp(timestamp)
      .build()
      .toByteArray
    case _          => throw new IllegalArgumentException("unknown type " + obj.getClass)
  }

  def fromBinary(bytes: Array[Byte], manifest: Option[Class[_]]): AnyRef = {
    val proto = ProtoChatMessage.parseFrom(bytes)
    ChatMessage(proto.getAuthor, proto.getContent, proto.getTimestamp)
  }

}

class FixedTimeMessageBuffer(duration: Long) extends Traversable[ChatMessage] {
  val list = ListBuffer[ChatMessage]()

  def now = System.currentTimeMillis()

  def old = now - duration

  def append(elem: ChatMessage) = {
    if (elem.timestamp > old) {
      while (list.nonEmpty && list.head.timestamp < old) {
        list.remove(0)
      }
      list.append(elem)
    }
  }

  override def toList = list.toList

  def replace(newList: List[ChatMessage]) = {
    list.clear()
    list ++= newList
  }

  def foreach[U](f: ChatMessage => U) = list.foreach(f)

}

...
sendIdentifyRequest()
 
def sendIdentifyRequest(): Unit = {
  log.info(s"Trying connecting to $roomReaderPath")
  context.actorSelection(roomReaderPath) ! Identify(roomReaderPath)
  context.system.scheduler.scheduleOnce(3.seconds, self, ReceiveTimeout)
}
 
def receiveRecover = Actor.emptyBehavior
 
def receiveCommand = identifying
 
def identifying: Receive = {
  case msg: ChatMessage => persistAsync(msg) { m =>
    log.info(s"Message $m persisted, but the reader isn't available")
  }
  case ActorIdentity(`roomReaderPath`, Some(actor)) =>
    log.info(s"Successfully connected to $roomReaderPath")
    context.watch(actor)
    context.become(active(actor))
  case ActorIdentity(`roomReaderPath`, None) => log.info(s"Remote actor is not available: $roomReaderPath")
  case ReceiveTimeout              => sendIdentifyRequest()
  case _                           => log.info("Not ready yet")
}
 
def active(reader: ActorRef): Receive = {
  case msg: ChatMessage => persistAsync(msg) { _ =>
    reader ! Update
    }
 
  case "snap" => saveSnapshot("foo")
 
  case Terminated(`reader`) =>
    log.info("reader terminated")
    sendIdentifyRequest()
    context.become(identifying)
  case ReceiveTimeout =>
  // ignore
}
...

class BackendTalker(roomWriterPath: String, roomReaderPath: String) extends Actor with ActorLogging {
  import BackendTalker._
 
  val listeners = collection.mutable.Set[ActorRef]()
 
  sendReaderIdentifyRequest()
  sendWriterIdentifyRequest()
 
  def sendReaderIdentifyRequest(): Unit = {
    log.info("sending identify request to reader")
    context.actorSelection(roomReaderPath) ! Identify(roomReaderPath)
    import context.dispatcher
    context.system.scheduler.scheduleOnce(3.seconds, self, ReaderReceiveTimeout)
  }
 
  def sendWriterIdentifyRequest(): Unit = {
    log.info("sending identify request to writer")
    context.actorSelection(roomWriterPath) ! Identify(roomWriterPath)
    import context.dispatcher
    context.system.scheduler.scheduleOnce(3.seconds, self, WriterReceiveTimeout)
  }
 
  def receive = identifying
 
  def identifying: Receive = {
    case ActorIdentity(`roomWriterPath`, Some(actor)) =>
      log.info(s"Successfully identified writer at $roomWriterPath")
      context.watch(actor)
      context.become(waitingForReader(actor))
 
    case ActorIdentity(`roomReaderPath`, Some(actor)) =>
      log.info(s"Successfully identified reader at $roomReaderPath")
      listeners.foreach(actor ! Listen(_))
      context.watch(actor)
      context.become(waitingForWriter(actor))
 
    case ActorIdentity(path, None) => log.info(s"Remote actor not available: $path")
 
    case ReaderReceiveTimeout =>
      sendReaderIdentifyRequest()
    case WriterReceiveTimeout =>
      sendWriterIdentifyRequest()
 
    case msg: ChatMessage =>
      listeners += context.watch(sender())
      sender() ! ChatMessage("system", "Connection problem, try again later", System.currentTimeMillis())
    case Terminated(userCon) => listeners -= userCon
    case _                           => log.info("Not ready yet")
  }
 
  def waitingForReader(writer: ActorRef): Receive = {
    case ActorIdentity(`roomReaderPath`, Some(reader)) =>
      log.info(s"Successfully identified reader at $roomReaderPath")
      listeners.foreach(reader ! Listen(_))
      context.watch(reader)
      context.become(active(reader, writer))
 
    case ActorIdentity(`roomReaderPath`, None) => log.info(s"Reader actor not available: $roomReaderPath")
    case ReaderReceiveTimeout =>
      sendReaderIdentifyRequest()
    case WriterReceiveTimeout =>
      sendWriterIdentifyRequest()
    case Terminated(`writer`) =>
      log.info("writer terminated")
      sendWriterIdentifyRequest()
      context.become(identifying)
 
    case msg: ChatMessage =>
      listeners += context.watch(sender())
      sender() ! ChatMessage("system", "Connection problem, try again later", System.currentTimeMillis())
    case Terminated(userCon) => listeners -= userCon
    case _ => log.info("Not ready yet")
  }
 
  def waitingForWriter(reader: ActorRef): Receive = {
    case ActorIdentity(`roomWriterPath`, Some(writer)) =>
      log.info(s"Successfully identified writer at $roomWriterPath")
      context.watch(writer)
      context.become(active(reader, writer))
 
    case ActorIdentity(`roomWriterPath`, None) => log.info(s"Reader actor not available: $roomWriterPath")
    case ReaderReceiveTimeout =>
      sendReaderIdentifyRequest()
    case WriterReceiveTimeout =>
      sendWriterIdentifyRequest()
    case Terminated(`reader`) =>
      log.info("reader terminated")
      sendReaderIdentifyRequest()
      context.become(identifying)
 
    case msg: ChatMessage =>
      listeners += context.watch(sender())
      sender() ! ChatMessage("system", "Connection problem, try again later", System.currentTimeMillis())
    case Terminated(userCon) => listeners -= userCon
    case _ => log.info("Not ready yet")
  }
 
  def active(reader: ActorRef, writer: ActorRef): Receive = {
    case l: Listen => reader ! l
    case msg: ChatMessage => writer ! msg
 
    case Terminated(`reader`) =>
      log.info("reader terminated")
      sendReaderIdentifyRequest()
      context.become(waitingForReader(writer))
    case Terminated(`writer`) =>
      log.info("writer terminated")
      sendWriterIdentifyRequest()
      context.become(waitingForWriter(reader))
    case ReaderReceiveTimeout =>
    case WriterReceiveTimeout =>
    // ignore
  }
 
}

context.system.scheduler.schedule(tenMinutes, tenMinutes, self, Snap)
val state = FixedTimeMessageBuffer(tenMinutes)
def updateState(msg: ChatMessage) = state.append(msg)

case msg:ChatMessage =>
      updateState(msg)
      sendAll(msg)
case Listen(ref) =>
      listeners add context.watch(ref)
      state.foreach(ref ! _)
case Snap => saveSnapshot(state.toList)
case SnapshotOffer(_, snapshot: List[ChatMessage]) => state.replace(snapshot)