When creating a subscription to an Azure Service Bus Topic you can add a filter which will determine which messages to send to the subscription based upon the properties of the message.
This is done by passing a SqlFilter to the Create Subscription method
e.g.
if (!_NamespaceManager.SubscriptionExists(topic, subscription))
{
if (!String.IsNullOrEmpty(filter))
{
SqlFilter strFilter = new SqlFilter(filter);
await _NamespaceManager.CreateSubscriptionAsync(topic, subscription, strFilter);
bSuccess = true;
}
else
{
await _NamespaceManager.CreateSubscriptionAsync(topic, subscription);
bSuccess = true;
}
}
Where strFilter is a string representing the properties that you want to filter on e.g.
// Create a "LowMessages" filtered subscription.
SqlFilter lowMessagesFilter = new SqlFilter("MessageNumber <= 3");
namespaceManager.CreateSubscription("TestTopic","LowMessages",lowMessagesFilter);
Applying properties to messages makes it easier to configure multiple subscribers to sets of messages rather than having multiple subscribers that receive all the messages, providing you with a flexible approach to building your messaging applications.
Subscriptions are effectively individual queues that each subscriber uses to hold the messages that a relevant to the subscriptions
When a message is pushed onto a Topic the Service Bus will look at all the subscriptions for the Topic and determine which messages are relevant to the subscription. If it is relevant then the subscription will receive the message into its queue. If there are no subscriptions capable of receiving the message then the message will be lost unless the topic is configured to throw an exception when there are no subscriptions to receive the message.
This approach is useful if most of the message data is stored in the properties (which are subject to a size limit of 64KB) and the body content is serialised to the same object (or the body object types are known).
Receiving messages on a Service Bus Subscription is as follows:
MessagingFactory messageFactory = MessagingFactory.CreateFromConnectionString(_ConnectionString);
SubscriptionClient client = messageFactory.CreateSubscriptionClient(topic, subscription);
message = await client.ReceiveAsync(new TimeSpan(0, 5, 0));
if (message != null)
{
properties = message.Properties;
body = message.GetBody<MyCustomBodyData>();
if (processMessage != null)
{
// do some work
}
message.Complete();
}
Over the past few months I have been looking at RabbitMQ and trying to apply my Service Bus knowledge, as well as looking at the differences. Routing messages based upon the message properties rather than a routing key defined in the message is still applicable in the RabbitMQ world and RabbitMQ is configurable enough to work in this way. RabbitMQ requires more configuration than Service Bus but there is a mechanism called Header Exchange which can be used to route messages based upon the properties of the message.
The first thing to do is to create the exchange, then assign a queue to it based upon a set of filter criteria. I’ve been creating my exchanges with an alternate exchange to allow me to receive message that are not handled in a default queue. The code to create the exchange and queue that subscribes to messages where the ClientId property is “Client1” and the FileType property is “transaction”.
// Create Header Exchange with alternate-exchange
IDictionary<String, Object> args4 = new Dictionary<String, Object>();
args4.Add("alternate-exchange", alternateExchangeNameForHeaderExchange);
channel.ExchangeDeclare(HeaderExchangeName, "headers", true, false, args4);
channel.ExchangeDeclare(alternateExchangeNameForHeaderExchange, "fanout");
//Queue for Header Exchange Client1 & transaction
Dictionary<string, object> bindingArgs = new Dictionary<string, object>();
bindingArgs.Add("x-match", "all"); //any or all
bindingArgs.Add("ClientId", "Client1");
bindingArgs.Add("FileType", "transaction");
channel.QueueDeclare(HeaderQueueName, true, false, false, args5);
channel.QueueBind(HeaderQueueName, HeaderExchangeName, "", bindingArgs);
//queue for Header Exchange alternate exchange (all other)
channel.QueueDeclare(unroutedMessagesQueueNameForHeaderExchange, true, false, false, null);
channel.QueueBind(unroutedMessagesQueueNameForHeaderExchange, alternateExchangeNameForHeaderExchange, "");
This will setup the exchange and queue in RabbitMQ and now you can send a message to the exchange with the correct properties as follows:
IBasicProperties properties = channel.CreateBasicProperties();
properties.Headers = new Dictionary<string, object>();
properties.Headers.Add("ClientId", "Client1");
properties.Headers.Add("FileType", "transaction");
string routingkey = "header.key";
var message = "Hello World";
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: TopicName,
routingKey: routingkey,
basicProperties: properties,
body: body);
Receiving messages from the queue is as follows:
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
var routingKey = ea.RoutingKey;
Byte[] FileTypeBytes = (Byte[])ea.BasicProperties.Headers["FileType"];
Byte[] ClientIDBytes = (Byte[])ea.BasicProperties.Headers["ClientId"];
string FileType = System.Text.Encoding.ASCII.GetString(FileTypeBytes);
string ClientID = System.Text.Encoding.ASCII.GetString(ClientIDBytes);
Console.WriteLine(" [x] Received '{0}':'{1}' [{2}] [{3}]",
routingKey,
message,
ClientID,
FileType);
EventingBasicConsumer c = model as EventingBasicConsumer;
if (c != null)
{
c.Model.BasicAck(ea.DeliveryTag, false);
Console.WriteLine(" [x] Received {0} rk {1} ex {2} ct {3}", message, ea.RoutingKey, ea.Exchange, ea.ConsumerTag);
}
};
channel.BasicConsume(queue: queueProcessorBaseName + textBox1.Text,
noAck: false,
consumer: consumer);
Again an out of the box feature for Service Bus can also be implemented in RabbitMQ but it is much simpler to use in Service Bus. The use of properties to route messages offers a much more flexible approach but does require that the body of the messages are either not used or are understood by each consumer. Service Bus offers more flexibility as the query string can contain a variety of operators whereas RabbitMQ matches all or some of the header values and not a range.