The AMQP actor#
The new-style actor that CLU introduces uses the AMQP protocol, of which RabbitMQ is the best known implementation, to communicate with other actors and clients. CLU supports AMQP 0-9-1; AMQP 1.0 is not supported at the moment.
In the following sections we’ll delve deeper into how the protocol is implemented and some of the code gritty details.
Why a new messaging protocol for SDSS?#
Say whatever you want about it, the current SDSS message passing protocol based on Tron, opscore, and actorcore is stable and robust. So, why should we replace it? Here is a list of reasons:
It reinvents the wheel. Ok, in all honesty
Tronandopscorewere written when wheel were still not completely circular, but the truth is that nowadays there are more robust, standard, and better documented technologies out there for message passing.We can remove the need for a central hub product by relying in open-source message brokers such as RabbitMQ.
Tronandopscoreare Python 2 and it’s unclear the amount of effort that would be needed to convert them to Python 3.While there is some documentation for
Tronandopscore, and the code is well written, it’s also cumbersome and difficult to modify by people that didn’t write it. It’s ultimately non-maintainable.The
opsctore/actorkeysdatamodel is custom-built and extremely difficult to maintain. Standard solutions such as JSON with a JSON schema validator should be preferred.asyncio provides an asynchronous API that is cleaner and easier to code than using threads. It is also more readable and less convoluted than twisted and it’s a Python core library with very active development.
CLU uses click for parsing commands, providing a well-defined, easy to use parser.
How does the new actor work?#
To completely understand how the code works you need a basic grasp of the AMQP protocol. The RabbitMQ tutorial is a good starting point, but let’s quickly go over some of the main concepts:
A program that sends messages is a producer. A message is any form of communication that can be encoded as bytes, for example a plaintext string or a JSON object.
A program that receives a message is a consumer. Actors are both producers and consumers.
Messages are sent to exchanges. Exchanges can be of different types, which enables various message brokering behaviours.
A queue is a buffer that holds messages. Queues subscribe to exchanges to get messages from them.
With all of that in mind, here is a conceptual diagram of how communication happens between two actors:
This may look daunting so let’s break it into smaller pieces:
We have two actors that are both producers (i.e., can send messages) and consumers (can receive messages). Messages can either be commands or replies.
There is a global exchange,
sdss_exchangeused both to send commands and replies. The exchange is of type topic, which allow queues to subscribe to them and receive only messages that match a certain routing key.Each actor has a queue for commands and another for replies. The command queue (
<actor>_commands) only gets messages whose routing key contains the keycommandand the name of the actor, and ignores all other command messages. The replies queue,<actor>_replies, receives all messages with topicreplyin the routing key.
More actors can be added to this picture, but they all share the same basic components. This enables us to create a slightly modified RCP pattern in which the callback queue is fixed and we use a topic exchange instead of a direct one.
The protocol#
Here we will formalise the AMQP actor protocol. This serves two purposes: document how CLU works internally, and enable other base actor classes to be developed (e.g., a thread-based actor).
Message brokering#
We use the AMQP standard for message queueing. The preferred implementation of the standard is RabbitMQ but any other broker that complies with the AMQP standard is acceptable.
Communication happens through a single topic exchange called sdss_exchange. Actors and clients are both consumers and producers. There is not formal difference between an actor and a client (e.g., a command line interpreter or a GUI) although the latter is not expected to receive commands, only emit them.
Each actor receives messages from two queues:
The commands queue must be called
<actor>_commands, where<actor>is the name of the actor. It subscribes to the exchange and listens to messages withcommandand<actor>in their routing key. Clients that are not actors do not need to have a commands queue.The replies queue is called
<actor>_replies. It subscribes to exchange and listens to message with areplyrouting key.
Queues must be exclusive and both exchanges and queues must auto-delete when the connection is closed.
Commands#
A command is a message that must conform to the following parameters:
Its routing key is
command.<consumer>where<consumer>is the name of the actor that will receive and execute the command. For example, ifactor1is commandingactor2, the routing key would becommand.actor2.The header of the message must contain two keys:
commander_idset to the name of the actor sending the command, andcommand_idset to a unique identifier for the command. UUID4 is recommended for command IDs.The
correlation_idof the message must be set to the same value ascommand_id.The content type of the message must be
text/json.The body of the message must be a JSON string with a simple key called
command_stringwhose value is the command string to be parsed by the remote actor.
Let’s guess that actor1 wants to send the command status --verbose to actor2 and that we create a unique identifier command_id='7b93d8d5-11c1-4c08-82a8-56842e1a86c4' for it. The information to be sent is
{headers: {command_id: '7b93d8d5-11c1-4c08-82a8-56842e1a86c4',
commander_id: 'actor1'},
payload: {command_string: 'status --verbose'},
content-type: 'text/json',
correlation_id: '7b93d8d5-11c1-4c08-82a8-56842e1a86c4',
routing_key: 'command.actor2'}
Replies#
The format for a reply to a command is as follows:
The routing key must be
reply.<producer>where<producer>is the actor or client that sent the command we are replying to. If an actor wants to reply with a broadcast to all the actors connected to the exchanged, it may do so with the routing keyreply.broadcast. This is technically not necessary since all actors and clients receive all message with topicreply, regardless of whether they have a secondary topic, but it allows finer filtering.The header of the message must contain the keywords
command_idset to the command ID of the command we are replying to,commander_idset to the name of the commander actor,senderset to the name of the actor replying, andmessage_codewith the message type code, which must be one of these.commander_idandcommand_idcan be null, if the reply is unrequested (e.g., a status command that runs on a timer).The
correlation_idmust be the same as thecommand_id(Nonefor broadcasts).The content type of the message must be
text/json.The body of the message must be a JSON string with a series of keyword-value pairs that conform to the keyword model of the actor that is replying.
In our example above, we commanded actor2 with status --verbose. Let’s imagine that the actor run that command and now wants to reply indicating that the lamps are on and the flat field screen is closed. The reply would look like
{headers: {message_code: ':',
command_id: '7b93d8d5-11c1-4c08-82a8-56842e1a86c4',
commander_id: 'actor1',
sender: 'actor2'},
payload: {lamps_on: true,
ffs: 'closed'},
content-type: 'text/json',
correlation_id: '7b93d8d5-11c1-4c08-82a8-56842e1a86c4',
routing_key: 'reply.actor1'}
The internals#
In this section we’ll have a quick look at how CLU implements the protocol we just defined. This is probably only useful to know if you are planning to override the AMQPActor class significantly.
For handling the connection to the RabbitMQ server while allowing asynchronous programming we use aio_pika, a wrapper around the Pika with support for asyncio.
When an AMQPActor is instantiated and we await start, a new instance of TopicListener is created, connects to the RabbitMQ server, and creates a channel and the topic exchange (or connects to it, if it already exists). We then add the commands_<actor> and replies_<actor> queues. When a new message is received by the command queue a callback to new_command is scheduled. When a new reply is received, we deal with it in handle_reply and use it to update the internal keyword model.
When a new message is received by new_command it unpacks the command ID and command string and creates a new instance of Command which is then passed to parse_command. This method is the one that invokes the click parser with the command string and calls the command function.
Internally, the top-level class in BaseClient, which defines the basic functionality for an actor/client (configuration parsing, logging, etc). The AMQPClient class implements a client based on an AMQP connection. The client provides a connection to the exchange, a replies queue, and methods to handle replies and issue commands. BaseActor expands BaseClient with a command parser and placeholders for handling incoming commands and writing but to the users. Both AMQPActor and LegacyActor subclass from BaseActor and implement their own specific protocols. AMQPActor also subclasses from AMQPClient given that reply parsing and command issuing are similar. The following diagram shows the inheritance tree: