Message Queues are very crucial part of many applications specially Enterprise and Real Time applications. There are many message broker frameworks available in the market e.g. RabbitMq, ZeroMq, ServiceBus and so on. Each framework has different features, implementation and client(s) to interact with the framework (more on message brokers).
It is very common that in a single application you use more than one message broker. For example, ZeroMq is in-memory messaging queue and it is very efficient because of this reason; so, it can be used for logging purpose in a separate thread. Or the scenario could be using free message broker in Development Environment and paid one in Production Environment. So there are many possibilities using more than one message broker in a single application.
The challenge is that you need to learn and maintain different clients, handle seriliazation (binary, json etc.), exception handling, logging, thread safety and also need to define some abstraction to easily switch between different message brokers. So you spend good amount of time on HOW TO DO rather than WHAT TO DO. This is the reason Dynamic Queue is born for. You just focus on the main stream of work and the rest is on Dynamic Queue. Sounds good! Right? Please continue reading....
Dynamic Queue (this framework) is an abstraction which is independent from any specific message broker implementation, hides the details of any message broker client, seamless serilization, multi-threading, common interfaces to interact and the flexibility to switch the message broker without updating and compiling the single line of code (yes, this is true).
We can generally define the communication patterns as follow:
- Fire and forget [FaF] (Push-Pull, Producer-Consumer)
- Request and response [RaR] (Remote procedure call, Server-Client)
- Publisher and subscriber [PaS]
Dynamic Queue framework defines abstraction around these communication patterns. It further divides the interfaces as inbound and outbound. Please see the below table for details:
| Pattern | Inbound-Interface | Outbound-Interface | Async |
|---|---|---|---|
| FaF | IInboundFaFMq<TMessage> |
IOutboundFaFMq<TMessage> |
✅ |
| RaR | IInboundRaRMq<TRequest, TResponse> |
IOutboundRaRMq<TResponse, TRequest> |
❌ |
| PaS | Not Implemented ❌ | Not Implemented ❌ | ❌ |
Dynamic Queue framework architecture has been designed with loosely coupled modules, interface-based dependencies, flexible configuration, seamless serialization and easy to extend. Below are further details:
The framework has built using C# 6.0 and .net 4.5.2.
All the exceptions from the framework are thrown as QueueException with error code defined in QueueErrorCode enum and message.
Logging is optional and if logger is passed while creating the instance of any type (inbound or outbound), then the logging will be done. There are two interfaces define in MessageQueue.Log.Core dll as IQueueLogger and IQueueLoggerAsync. The default logger using NLog is defined in MessageQueue.Log.NLog and the usage can be seen in Samples (please see the Samples section below).
Serialization is seamless and is done using Newtonsoft.
The configuration can be stored in any type of store or configuration file. There is an interface named as IQueueConfigurationProvider in MessageQueue.CofigurationProvider.Core dll and the default implementation which retrieves configuration from AppSettings is defined in MessageQueue.CofigurationProvider.AppSettings. If you want to define your custom configuration provider, then simply implement IQueueConfigurationProvider.
<!-- ZeroMq -->
<add key="ZeroMqFaFOutbound:Address" value=">tcp://localhost:5551" />
<add key="ZeroMqFaFOutbound:Implementation" value="MessageQueue.ZeroMq.Concrete.Outbound.ZmqOutboundFaF`1, MessageQueue.ZeroMq" />
<!-- RabbitMq -->
<add key="RabbitMqFaFOutbound:UserName" value="guest" />
<add key="RabbitMqFaFOutbound:Password" value="guest" />
<add key="RabbitMqFaFOutbound:Address" value="localhost" />
<add key="RabbitMqFaFOutbound:QueueName" value="Test_SampleQueue" />
<add key="RabbitMqFaFOutbound:Implementation" value="MessageQueue.RabbitMq.Concrete.Outbound.RmqOutboundFaF`1, MessageQueue.RabbitMq" />
/>RabbitMqFaFOutbound and ZeroMqFaFOutbound are configuration identifiers to group the configuration for any particular instance.
All the implementations of message brokers (some by default from the message broker and others managed by Dynamic Queue) are thread safe.
There is a class named as MessagingQueueFactory which is responsible to create any kind of interface implementation. This class is available in MessageQueue.Core dll. It exposes static methods for instance creation. It takes configuration provider, configuration identifier (string literal which groups the configuration) and optionally logger. Please see Samples for details.
Creation of any interface implementation is dyamic and is based on the fully qualified class name and assembly name.
As of now, following message brokers have been implemented:
| Message Broker | FaF | RaR | PaS |
|---|---|---|---|
| ZeroMq | ✅ | ✅ | ❌ |
| RabbitMq | ✅ | ✅ | ❌ |
| ServiceBus | ✅ | ❌ | ❌ |
See message brokers comparison
| Name | Description | Example | Required |
|---|---|---|---|
| Address | The address of the queue (no server as it is in-memory) | >tcp://localhost:5551 |
✅ |
| Implementation | The relevant implementation (inbound or outbound) | MessageQueue.ZeroMq... |
✅ |
| Pattern | Inbound-Interface | Outbound-Interface |
|---|---|---|
| FaF | MessageQueue.ZeroMq.Concrete.Inbound.ZmqInboundFaF`1, MessageQueue.ZeroMq |
MessageQueue.ZeroMq.Concrete.Outbound.ZmqOutboundFaF`1, MessageQueue.ZeroMq |
| RaR | MessageQueue.ZeroMq.Concrete.Inbound.ZmqInboundRaR`2, MessageQueue.ZeroMq |
MessageQueue.ZeroMq.Concrete.Outbound.ZmqOutboundRaR`2, MessageQueue.ZeroMq |
| Name | Description | Example | Required |
|---|---|---|---|
| Address | The address of the RabbitMq server | localhost |
✅ |
| Implementation | The relevant implementation (inbound or outbound) | MessageQueue.RabbitMq... |
✅ |
| QueueName | The queue name | MyQueue |
✅ |
| UserName | Username to connect with server | guest |
✅ |
| Password | Password to connect with server | guest |
✅ |
| Port | The port on which server is listening | 1234 |
❌ |
| Acknowledgment | The message acknowledgment setting (inbound only) | true OR false |
❌ |
| MaxConcurrentReceiveCallback | The max number of concurrent calls to the receive handler | 5 |
❌ |
| ExchangeName | The exchange name | MyExchange |
❌ |
| RoutingKey | The routing key | Key1 |
❌ |
| ConnectionTimeoutInMinutes | The connection timeout in minutes | 2 |
❌ |
| Pattern | Inbound-Interface | Outbound-Interface |
|---|---|---|
| FaF | MessageQueue.RabbitMq.Concrete.Inbound.RmqInboundFaF`1, MessageQueue.RabbitMq |
MessageQueue.RabbitMq.Concrete.Outbound.RmqOutboundFaF`1, MessageQueue.RabbitMq |
| RaR | MessageQueue.RabbitMq.Concrete.Inbound.RmqInboundRaR`2, MessageQueue.RabbitMq |
MessageQueue.RabbitMq.Concrete.Outbound.RmqOutboundRaR`2, MessageQueue.RabbitMq |
| Name | Description | Example | Required |
|---|---|---|---|
| Address | The servicebus endpoint address | Endpoint |
✅ |
| Implementation | The relevant implementation (inbound or outbound) | MessageQueue.ServiceBus... |
✅ |
| QueueName | The queue name | MyQueue |
✅ |
| Acknowledgment | The message acknowledgment setting (inbound only) | true OR false |
❌ |
| MaxConcurrentReceiveCallback | The max number of concurrent calls to the receive handler | 5 |
❌ |
| Pattern | Inbound-Interface | Outbound-Interface |
|---|---|---|
| FaF | MessageQueue.ServiceBus.Concrete.Inbound.SbInboundFaF`1, MessageQueue.ServiceBus |
MessageQueue.ServiceBus.Concrete.Outbound.SbOutboundFaF`1, MessageQueue.ServiceBus |
| RaR | Not Implemented | Not Implemented |
The framework does not create the queues. The only scenario is in RaR pattern where it needs to create queue for the responses.
If you want to run the code in Visual Studio or any other .Net IDE, just download the source code, restore the nuget packages, update the configuration and you are good to go. Please see the section Samples below for details.
In the Test folder, there are four projects (console application) which consumes Dynamic Queue for each supported communication pattern and message broker. For FaF patter, MessageQueue.Sender [ code snippet ] and MessageQueue.Receiver[ code snippet ] test projects are configured and for RaR pattern, MessageQueue.RaR.Server [ code snippet ] and MessageQueue.RaR.Client [ code snippet ] test projects are configured. Configuration for all the projects is stored in AppSettings.cofig file.
| ZeroMq | RabbitMq | ServiceBus |
|---|---|---|
| ✅ | ✅ | ✅ |
| ZeroMq | RabbitMq | ServiceBus |
|---|---|---|
| ✅ | ✅ | ❌ Not Implemented |
Like it? Wants to apricate? Please go ahead!!!
