Multi Response pattern using RabbitMQ

A while ago I was experimenting with distributed architectures, micro services and scalability. I like the idea of micro services that are accessible via centralized queue technology. Main benefits of such an architecture are:

• scalability in production by the self-balancing nature of queues
  Just spin up more instances of (micro) services that listen to the same work queue for some additional processing power / throughput.
• agility in development
  Microservices are one-trick pony's. That's why they tend to have a relatively small code base, which make them intrinsically more maintainable.

Scenario

In my scenario, multiple sources should be queried and the results need to be combined and processed together on the client. Each service has its own implementation details for returning the source data.

The client is agnostic to the services. So the client does not know upfront how many services will respond to its request message. Services will decide for itself whether a certain request is worthwhile to respond to.

I wondered how I would implement a communication pattern like that.
And I would like to implement it using my knowledge of .NET, RabbitMQ, AMQP and as much EasyNetQ as I possibly can use.
The communication can be signatured like this:

public interface IRequestMultiReplyClient<TRequest, TResponse>
{
   Task<IEnumerable<TResponse>> SendAsync( TRequest request );
} 

I will first describe the conceptual protocol design and some of the problems to overcome.

Then I will refine the protocol design to overcome the problems.
In part 2 of this blog post,  I will go into the implementation details using EasyNetQ.

Note: A somewhat related and similar pattern is described by the author of EasyNetQ (Mike Hadlow) in his post about the Request-Response pattern. Also some of the problems he describes are also applicable to Request-MultiResponse.

Conceptual design

Request-Multi Response is in basis just a combination of two well-known messaging patterns: Publish-Subscribe for the request & Send-Receive for the responses. 

Initial communication setup

• The services subscribe to an exchange.
  Subscribing means:
• Creating 1 exchange (if not existent)
• creating a private queue for each server 
• binding the private queue to the exchange. 
• listening on the private queue for incoming messages.
• The client creates a response queue  

Basic interaction

• The client starts listening on the response queue for incoming response messages.
• The client publishes a message to the exchange. 
• A service will receive the request message and will start processing the message.
• A service sends a response message to the response queue
• The client receives the response message.
• The client stops listening "after a while".

Problems: Concurrency & Isolation

You might already spotted some problems and limitations with the current design.

• Are you creating a response queue for each request? That's pretty costly...
• How does the server know to which response queue to send the response?
• How does the client know how much responses will be received? 
• How long does the client have to wait on the responses?
• Can you send different sorts of message types?

Refinements

Exclsuive reply queue per client

Let's refine the solution and just use an exclusive reply queue per client to reduce the overhead.
We will use a default property of AMQP message named 'reply-to' to transport the queue name to the other end. This property will be used to instruct the server to which queue to send the response.The client is sending a message by PubSub with the value of the reply-to property set to the name of the particular response queue for the client.
This implies that the client should have logic to correlate an incoming response to the correct request.

Correlate using correlationId

For correlating messages we can make use of standard features in the AMQP protocol.
The default properties of an AMQP message has a CorrelationId field. When we are sending in the request message over PubSub we will ship it with a generated CorrelationId. The server will use the same correlationId to send the response over the Send-Receive channel. The client will receive the message and is able to correlate it with the initial request.

Completion signaling

Next to a timeout, let's introduce a second form of completing the communication by using an intermediate message.This type of message is a response which tells the client that a specific service is busy processing the request. As a result a service will send 2 responses for each received request: an intermediate message and finally a complete message.
On the client side, we split the receiving process into 2 phases and keep an administration of all incoming responses.

First Receive Phase
Intermediate responses received in the the first receive phase will determine for which complete responses to wait in second phase. This first phase will be finished after a specified amount of time.
A complete response received in this first phase will be administered as a final response.

Second Receive Phase
In the second phase we receive complete responses and with each complete response we will remove the related intermediate response (using the correlationId).
This phase will be finished when all intermediates are removed OR the timeout period has been expired.
Intermediate responses received in the second phase are being ignored.
A complete response without a related intermediate is treated as a valid complete response in this phase.

Multiple response administrations
To be able to deal with multiple concurrent requests, the response administration should be unique for each request. For each received response message we will first look up the correct response administration.
The response message should contain:

• CorrelationId: an identifier of the initial request to be able to find the correct response administration
• ResponseId: an identifier to be able to pair intermediate & complete responses
• IsIntermediate: boolean to indicate whether the message is a intermediate or complete response.

In next post I will show you the implementation using EasyNetQ.