Kino: communication frawemork on NetMQ. Short description

About 8 years ago I started working in a team that developed one service. The interface of the service was quite simple, only 4 methods, and it performed a single task. During all this time, the code was constantly changing: new business rules and restrictions were implemented, versioning was added. At one point, the front-end needed a very small functionality that was buried deep in service. The implementation of the necessary function was developed as a component and did not pose any problems to give access to it from the service through an additional method ... Except for one: the logical coherence of the service methods was violated, that is, its "insides" began to become "externals".

The problem could be solved if we transform all these small internal components, to which external access was required, into separate services. In this case, the front-end could gain access to their functionality; the main service would become more compact and its role was reduced to orchestration of calls.

We used WCF to build services. Deploying the service in 50 lines of code on WCF, at least on 3-4 servers, with load-balancer, new URLs and other bells and whistles, seemed not a good idea. And I wanted some kind of lightness, perspective ...

A few years later, I participated in another project at Workflow Foundation. Looking at what worked in the XAML editor, I thought: “Why not imagine the whole workflow as a sequence of messages”?

Kino tests

The search for existing solutions, to be honest, I did not. At that time (4-5 years ago) little was known about “Orleans” , and I learned about Akka after the start of cycling. On the one hand, this is bad, unworthy of a professional developer and all that. On the other hand, something new could have turned out ... How good or bad everything turned out, a respected reader can judge.

So, I started creating the kino : actor-like communication framework on NetMQ . The suffix "-like" because classical actors have a hierarchical organization, supervisors, they are stateful and, in general, they have a whole mathematical model there ... Everything is simpler, but, nevertheless, we will have actors too.

In short, what is what?

The main means of communication in kino is the message. Each message has a version and a type, which are used to find the corresponding handler. There is a slight deviation from the rule, but for now let's not talk about it.

Actors ( Actors ) - the main consumers and producers of messages. Actor announces its interface by specifying the type and version of the message that it can receive. There is another extras member, MessageHub , which can also receive and send messages. However, there are certain differences between them. Actor should be considered as a service: it can only respond when it receives an incoming message. MessageHub is a client that can send a message and (try) receive a reply message if necessary. So, most often, the initial message is sent via MessageHub and processed by one or more Actors .

To search for recipients of messages, MessageRouter is required. It stores the routing table — version matching (Version) and message type (Identity) with a list of Actors that can process it. For one process, one MessageRouter is enough.

To go beyond one process / host, we need to gain knowledge about the outside world, that is, about the other MessageRouters and their routing tables. The source for this knowledge is the Rendezvous service. This is the only well-known address that must be configured for the kino- based application. Rendezvous accepts from everyone and distributes to all connected MessageRouters the information about adding new and removing nonexistent routes, ping the established connections. Rendezvous service forms a single network of kino components.

Also, but in more detail

1. Message

This is a typical message that you can send a walk on the kino network:

public class MyMessage : Payload { private static readonly byte[] MessageIdentity = "NAMESPACE.MYMESSAGE".GetBytes(); private static readonly byte[] MessageVersion = "1.0".GetBytes(); //    , .. ,       public override byte[] Version => MessageVersion; public override byte[] Identity => MessageIdentity; } 

There are 3 distribution patterns (Distribution Pattern) supported: unicast , broadcast and direct . In the first case, the message is sent to only one processor registered in the network. In the second - all.

 IPayload payload = new MyMessage(); IMessage message = Message.Create(payload, DistributionPattern.Broadcast); 

In the case of direct distribution , which can be particularly useful in testing, the message is sent to a specific MessageRouter 's:

 IMessage message = Message.CreateFlowStartMessage(new MyMessage()); message.SetReceiverNode(receiverIdentity); //     

You can get to the data in the received message as follows:

 MyMessage payload = message.GetPayload<MyMessage>(); 

2. Actors

To create your actor, you must inherit the class from Actor and implement at least one message handler method in it:

 public class MyMessageProcessor : Actor { [MessageHandlerDefinition(typeof (MyMessage))] public async Task<IActorResult> MyMessageHandler(IMessage message) { //   } [MessageHandlerDefinition(typeof (OtherMessage))] public Task<IActorResult> OtherMessageHandler(IMessage message) { //   } } ```cs      ,       *kino*.         ,     : ```cs public class LocalMessageProcessor : Actor { //      [MessageHandlerDefinition(typeof (LocalMessage), true)] public async Task<IActorResult> MyMessageHandler(IMessage message) { //   } } 

The framework guarantees that the actor's handler method will receive only messages of the type that is declared. The result returned can be one or more messages of any type, and with different distribution patterns . That is, we can send a response to the initial sender and at the same time inform everyone else of something else:

 public class MyMessageProcessor : Actor { [MessageHandlerDefinition(typeof (MyMessage))] public async Task<IActorResult> MyMessageHandler(IMessage message) { MyMessage payload = message.GetPayload<MyMessage>(); var result = await UpdateDb(payload); IMessage response = Message.Create(new ResponseMessage(result)); IMessage notifyRequest = Message.Create(new NotifyMessage(result), DistributionPattern.Broadcast); return new ActionResult(response, notifyRequest); } } 

3. ActorHost

We haven't talked about ActorHost yet. This is a component that performs several functions:

• stores references to handler methods of all registered Actors
• register handlers in MessageRouter
• accepts messages for processing from MessageRouter to Actors and sends responses back to the MessageRouter .

Calling handler methods in ActorHost occurs in a single thread (with the exception of asynchronous methods). Therefore, ActorHost does not support multiple registrations of handlers of the same message. If it is necessary to scale the same type of Actor within the same process, it is necessary to create a new ActorHost instance for each of them. ActorHostManager assumes all these difficulties in choosing and creating ActorHosts :

 //     MyActor IActor actor = new MyActor(); //  ,    ActorHost actorHostManager.AssignActor(actor); //  ,    actor = new MyActor(); // ,       MyActor actorHostManager.AssignActor(actor, ActorHostInstancePolicy.AlwaysCreateNew); 

4. MessageHub

Let's go back a little, how it all began. And it began with the fact that there was a need to spread the code from one WCF-service to several components accessible on the network. As a result, instead of calling hundreds of methods in one process, we got a certain flow of messages ( message flow ), which, in addition, travel to different servers. However, the functionality and behavior for the end user of the service should ideally remain the same. That is, if the client used to call the service method synchronously and expect to receive a response, then with all this kino , the client's work pattern should not change drastically. It is necessary to determine from all this message flow what is the answer to the client and deliver it back.

MessageHub is designed to solve this problem. With it, you can send a message to the kino network, without waiting for an answer:

 IPayload payload = message.GetPayload<MyMessage>(); IMessage message = Message.CreateFlowStartMessage(payload); messageHub.SendOneWay(message); 

And you can also specify that the sender expects a specific answer:

 //  ,   IMessage request = Message.CreateFlowStartMessage(new StartMessage()); // ,         ICallbackPoint callbackPoint = CallbackPoint.Create<ResultMessage>(); //       using(IPromise promise = messageHub.EnqueueRequest(request, callbackPoint)) { if(promise.GetResponse().Wait(timeout)) { //    ResultMessage result = promise.GetResponse().Result.GetPayload<ResultMessage>(); } else { //  … } } 

5. MessageRouter

MessageRouter is a node on the kino network. It connects other components, ActorHosts and MessageHubs , for messaging. In turn, MessageRouters find their own kind and connect to each other using the Rendezvous service, thus forming a kino network.

The kino uses the NetMQ library as a transport. She was practically hammered into the framework with nails and it was not planned to use other transport.

So, message routing. It is carried out according to the following algorithms:

• Unicast message:

    ActorHost  MessageHub,             MessageRouter  ,              ! 

• Broadcast message:

     ActorHosts  MessageHubs,            broadcast-     Actor   MessageRouter  ,                      ! 

• Direct message:

  unicast-  MessageRouter  ,   ,   (ReceiverNode),              !  < broadcast-> 

6. Rendezvous

Rendezvous service is the only well-known service whose address must be configured for all nodes on the same kino network. It performs the following functions:

• broadcast — redirect routing changes: adding new routes and deleting invalid routes.
• broadcast - PING distribution of messages for monitoring connected nodes,
• broadcast — sends PONG reply messages from connected nodes.

If necessary, the Rendezvous service can be installed on a server cluster. The leader selected by consensus is responsible for all of the above functions. In the event of a "fall" of the cluster, the kino network will continue to work. However, information about changes in routing will not be available. When the Rendezvous service is restored, the nodes will receive a network configuration update.

Open questions

• Well, actually, to see something in the production. Until it came to that ...
• How to work with messages of different wire format in the same network
• Possible problems with a large number of connections to the Rendezvous service, batch processing of PONG messages
• Combining multiple kino networks, that is, routing between nodes connected to different Rendezvous servers / clusters

The kino project on Github: https://github.com/iiwaasnet/kino
Wiki: https://github.com/iiwaasnet/kino/wiki