リアクティブメッセンジャー、またはAkkaおよびScalaとCQRSおよびES

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)