Jet Messenger, or CQRS and ES with Akka and Scala

Recently, we often hear about reactive programming and see various basvords: message-driven architecture, event-sourcing, CQRS. Unfortunately, on Habré they write quite a bit about it, so I decided to correct the situation and share my knowledge with everyone.

In this article we will learn about the main features of reactive applications, consider how the CQRS and EventSourcing patterns will help us in their creation, and in order not to be bored, we will step by step create your messenger with a websocket and actors that matches all canons of reactive programming. To implement all this stuff, we will use the wonderful Scala language along with the equally excellent Akk library that implements the model of actors. Also, we will use the Play Framework to write the web part of our application. So let's get started.

The article is intended for those who are already familiar with Scala and heard about the model of actors. All others are also invited to read, the principles of reactive programming can be applied regardless of the language and framework.

What is reactive programming

The idea of ​​reactive programming is described in the reactive manifest www.reactivemanifesto.org . Translation of its first version was already on Habré, and the second version is slightly different from the first. Let's look at a brief clipping from the second version. A reactive manifest states that reactive applications have several important properties:
')

Responsiveness

The application responds as quickly as possible. Responsiveness is the basis of usability and utility, for the simple reason that long interface delays do not add a desire to use it, and also responsiveness means that problems can be quickly detected and effectively solved. Responsive systems focus on delivering fast and consistent response times, using suitable upper time boundaries to ensure consistent quality of service. This constant and predictable behavior, in turn, simplifies error handling, enhances user confidence, and encourages them to further interact.

fault tolerance

The application remains responsive when a failure occurs. This applies not only to high-availability, mission-critical systems — any failure-unstable system will not be responsive in the event of a failure. Stability is achieved through replication, localization, isolation and delegation. Failures do not go beyond the module, and by isolating the modules from each other, you can be sure that individual parts of the application can fail and recover from a failure, while not causing the entire application to fall. Recovery of each failed module is delegated to another, external module, and high availability is achieved through replication. Module clients do not have a headache with module failure handling.

Elasticity

The app remains responsive under various loads. Reactive applications can respond to changes in load by increasing or decreasing the resources intended for its processing. This implies an architecture that does not have locking points or central bottlenecks, which is reflected in the ability to sharding and replication of modules, and the further distribution of the load between them. The result of this is scalability with the help of cheap commonly used hardware (hello, google!).

Message Orientation

Reactive applications focus on asynchronous message passing to set boundaries between modules that provide weak connectivity, isolation, location transparency, and provides means for delegating errors as messages. Introducing explicit message passing provides opportunities for load balancing, elasticity, and flow control by generating and monitoring message queues, and reducing bandwidth when needed. Location transparency allows the same error handling logic to be used on the cluster as well as on the same node. Non-blocking communications allow message recipients to consume resources only when they are active, which leads to a smaller overhead monitor when an application is running.

CQRS

CQRS stands for Command Query Responsibility Segregation (division of responsibility into teams and requests). This approach to building the application architecture, in contrast to the widely used CRUD (Create Retrieve Update Delete), implies that it is possible to use different models for updating and reading information. A natural question arises, why do we need such perversions? The fact is that based on the fact that the reading model and the writing model are separated, we can optimize them for these tasks. For example, if data is better suited for reading tasks, then no one bothers us to do this. It is more convenient to read if the data in the graph database - please. I want to store everything in the Key-Value store - for God's sake. Moreover, if you want to add new features to the read model, then all you need to do after adding them is to regenerate the model (it’s worth making a reservation that if we have events for many gigabytes, then this process will not be so fast, but we can always make snapshot, which will significantly increase the speed of recovery).

In principle, over the normalization of the Read-model, you can not bother at all, for the same reason. Using CQRS to optimize read operations in our application, we ensure the responsiveness of our application. What do we still have in order for our application to be truly reactive? Correct, elasticity and resiliency. We implement these features using the Event Sourcing pattern.

Event sourcing

The meaning of ES is that we do not store the current state of our data model, but the entire history of changes that change the state of our application (in fact, not all changes, but only those that matter to us). To get the current state, we simply summarize the changes from all existing events. What do we mean by an event, and how does an event differ from a team? The team means that someone wants from us, besides, it can be ignored. An event is something that happened, an unchangeable fact.

The advantage of this approach is that we never delete or change anything. As you may have guessed, this gives us ample opportunities to scale our application, and as a database we can use well-proven NoSQL solutions such as Cassandra or HBase. EventSourcing gives us resiliency and elasticity.

Stop talking, show us the code.

So, as mentioned earlier, we will implement the whole thing using the Typesafe stack .

The architecture of our application will look like this:



The user has the ability to read and send messages. Sending and receiving messages occurs through the web socket, which the UserConnection actor has access to. This actor sends messages to the RoomWriter actor, which, in addition to writing messages to the log, is engaged in kicking the RoomReader actor, which reads messages from the log and sends them back to the UserConnection actor. In addition to all this, we have the Receptionist actor, which handles the issuance of names and ensures that the application does not have users with two identical names. C architecture more or less figured out, now begin to write code.

RoomWriter

The very first we implement the actor that records incoming messages in the log.


What is it written here? As you might guess, we declared the RoomWriter class, which has three parts:

• identifier persistenceId, which is necessary to uniquely identify the events that were produced by this actor;
• a set of listeners containing a set of links to actors that should be notified that something has changed in the log;
• two methods, receiveRecover, which is called when replaying messages from the log that occurs when creating an actor, and receiveCommand, which is used to process messages during normal operation.

Let's look at the receiveCommand method in more detail. This method handles three different messages:

• when a Message is received, it is asynchronously written to the log, and a message is sent to each listener that the log has been updated.
• when we receive a Listen, we begin to follow the life cycle of the actor, the link to which lies in the message, among other things, the link to the actor is added to the set of listeners
• A terminated message containing a link to the deceased actor will be received if the actor over the life cycle of which we are watching will suddenly die. If this happens (the user has closed the browser), then we remove this actor from the mailing list.

The rule of good tone is the declaration of all processed messages and the factory method for creating an actor in a companion object:


With RoomWriter, we figured out, now is the time to look at the RoomReader actor, which receives updates from the log, and sends them through the hierarchy above.

RoomReader

RoomReader depends on the log identifier, depending on which it will receive its updates. In our case, this identifier will match with the identifier of the actor RoomWriter, which will mean that everything that RoomWriter writes to the log will come to RoomReader. Consider how message processing takes place:

• when a Message is received, the time it is sent is checked, and if the message is older than ten minutes, it will not be shown to the user. This is done so that thousands of previously accumulated messages do not arrive to the user.
• when receiving Update, the actor reads the log, and sends the read messages to the user.

As in the previous case, our companion object:


We proceed to the most interesting part, the UserConnection actor, which is responsible for processing messages from the web socket.

Userconnection

This actor has one feature that distinguishes it from others: it can change its behavior and state. Initially, he is in a state of waiting for a username. In this state, it can accept client requests for a name, and forward them to the actor responsible for issuing names. Upon successful receipt of the name, the actor enters the state of readiness for the chat, and begins to send messages between parts of the system.

This companion turned out to be quite large this time:



The last actor to be given our attention is the Receptionist.

Receptionist

Its task is to issue names to users: it contains an associative array that maps names to actorRefs. Just like in RoomWriter, we follow the life cycle of the actors to whom we gave out the names, and in case of their death, remove their names from the list of registered names.

Do not forget about the companion object, we place the factory method there to create the actor:

Controller

At the moment we are finished with all the actors that we had plans for implementation. Now let's look at how we can connect a web socket and an actor. For this, we will use the tools that the play framework can offer us. We implement the controller of our application as follows:


First, we create two actors: roomWriter and receptionist. They are dependencies for the UserConnection actor. Next, we describe how to format messages for transmission through the webcast. Finally, we describe how we handle incoming connections to the web socket. The helper built into the Play Framework makes it incredibly easy to do.

It's time to create a web interface. For the layout we will use the twitter bootstrap framework, and angular.js - to implement the business logic on the client.


What our html page will look like:

Scaling out

We have a prototype of the application, but before rolling it out into production, we should pump it a little. We will pump it in the following way:

• Let's replace our ersatz magazine with something really good. In this case, we will take Cassandra, and will use it to store events.
• Default Java serialization does not differ in its stability when changing the format of messages or the speed in their serialization. It is worth replacing it with Google Protobuf or Kryo. In this case, we will use Protobuf.
• Users of our messenger want to stay up to date with the latest news, and do not want to read messages older than half an hour. To do this, we will change the logic of our actors, and we will create a snapshot every half hour, so we do not have to restore the entire message history every time a user connects.
• In order for the application to process a large number of users, it is worth making it distributed.

When you run an application on multiple servers, its architecture will change very slightly. Due to the fact that the actors in Akka have the property location transparency, we can safely pull away our application on several servers. Moreover, our actors will not even guess that they are now separated and work on different servers communicating over the network. All we need is to add some code, and Akka will do the rest of the work for us.
I'll run ahead and provide a picture of how our application will look after all the improvements. In general, the architecture will undergo minor changes, but the idea will remain the same.



To use cassandra as a magazine, we need

1. install cassandra on nodes,
2. use the plugin to keep the log in cassandra.

The first paragraph is well described in the official manual, so there is no special reason to bring it here. It is worth noting only that it is not necessary to make all the nodes of the cassanda seed-nodes, for a cluster of three machines one sid will be enough.

Regarding the second, we need to specify the type of log in the config, and register the addresses of the cassandra nodes. This can be done as follows:

After connecting cassandra, we will write our class to serialize and deserialize messages: first, we will use the protobuff code generator to generate the necessary classes, and then with their help we will make a serializer.

This is how the protobuf file will look like:


After the generation of the required class by the protobuf, we will write our serializer:


So, we now have a normal log, and a way to write to it. Now we need to think of a way to save messages no longer than 10 minutes. To do this, we write our own buffer that will save messages in the last 10 minutes.


We chose ListBuffer as the data structure for storing messages - for the reason that we only add elements to the end and remove them from the beginning. ListBuffer allows you to do these operations for a constant time. In the future, we will apply this buffer in our Reader actor in order to limit the number of messages sent to newly connected clients.

Consider how we divide actors across the network. In order for our application not to fall when one node is disconnected, and waiting for it to be turned on, we need to write the appropriate logic in the actor. Actor RoomWriter should notify RoomReader about new messages, so it will be useful for him to know the state of RoomReader. This logic is well described by the introduction of two states in the actor.


In the sendIdentifyRequest method, we try to get the ActorRef of the remote actor by sending it a Identify message. All actors understand this message, and in response they send us the ActorRef we need. After receiving the ActorRef, we go back to normal and begin work. It should be noted that we are also starting to monitor the life cycle of the remote actor and, if it is not available, we will try to reach it again.

In order to implement such a logic of work for the UserConnection actor, we will create a separate actor that will act as an intermediary in communicating with the backend.


In it, we implement the logic of waiting for remote actors by analogy with what we did in the RoomWriter actor. In this case, we need to expect connection to two actors at once, so the logic of the work is a bit more complicated.

The final touch remains: we will rewrite RoomReader a bit in order to limit the number of messages that users receive.

To do this, we will add a couple of lines in it.

In the constructor, we define our buffer for storing messages, and write an auxiliary method for working with it. In addition, we will launch the scheduler, which every 10 minutes will give the command to create snapshots. It is worth noting that the command is given by sending a message to the actor, and we do not call the saveSnapshot method directly. This is done specifically in order not to violate the principle that the work with mutable data of the actor should be done only by the actor. Violating this principle, we can get subtle bugs.


In the receive method, we implement the ability to save snapshots after a special message arrives. We also implement the correct recovery of the state from snapshot.


Summarizing, we can say that we have implemented a modern web application, made in the spirit of reactive programming. It allows you to quickly respond to user requests and also has some degree of stability. However, it has much to improve. In order for our application to work even in the case of the fall of individual nodes, we should use the akka-cluster module, which allows you to quickly implement decentralized applications that do not have a single point of failure.In addition, we need to somehow handle the situation when the flow of messages is too large and the actors do not have time to process it. To work with this, there is an experimental module akka-streams. We will find out about this and many other things in the next article.