When you use Service Bus queues for an enterprise application, communication messages are picked up based the FIFO pattern. This means if a queue's length is 100, then a high-priority message will only be delivered after the subscribing application reads and commits all of the previous 100 messages.
To deal with this scenario, one option is to create separate queues for high-priority and low-priority messages, and the client application will be in control of setting up a queue endpoint when publishing a message to Service Bus queues:
The code for routing the message to separate queues is listed here. In this method, the client application needs to make different a Service Bus connection object to send messages to high-priority and low-priority queues:
public async Task SendMessageTopriorityQueue()
{
QueueClient highpriorityqueueClient = new QueueClient(sbconnection, "highpriorityqueue");
QueueClient lowpriorityqueueClient = new QueueClient(sbconnection, "lowpriorityqueue");
for (int i = 0; i < 10; i++)
{
string messageBody = $"priorityqueue Message {i}";
var message = new Message(Encoding.UTF8.GetBytes(messageBody));
await highpriorityqueueClient.SendAsync(message);
}
for (int i = 0; i < 10; i++)
{
string messageBody = $"lowpriorityqueue Message {i}";
var message = new Message(Encoding.UTF8.GetBytes(messageBody));
await lowpriorityqueueClient.SendAsync(message);
}
}
Similarly to the preceding code, when working with Logic Apps and Service Bus, you can select different priority queues that can act either as publishers or subscribers to the message. In the following example, we have used Service Bus connectors for queues in logic apps to pull out the sent message and route the message to the HTTP endpoint:
When working with logic apps and Service Bus connectors, you need to work with content type conversion from Base64 into a JSON string. This can be achieved through the Logic Apps expression language and the built-in base64ToString function:
"HTTP": {
"inputs": {
"body": "@{base64ToString(triggerBody()?['ContentData'])}",
"method": "POST",
"uri": "http://requestbin.fullcontact.com/15djfau1"
},
"runAfter": {},
"type": "Http"
}Another option is to use Service Bus topics, which can work as virtual priority queues with the Service Bus topic. There can be any number of applications subscribing to the same topic.
In the Service Bus topic approach, we need to pass additional user-defined properties along with a message body when publishing a message to the topic endpoint. Creating a priority topic with subscription uses the SQLfilter expression to differentiate between incoming messages.
To create a SQL filter rule for a priority, you can either use the Service Bus SDK, as shown in the preceding section, or you can use tools such as Service Bus Explorer or Serverless360. As this book focuses more on real implementations, we will use code to define the subscription filter condition for low-priority and high-priority messages:
public async Task CreatetopicprioritySubscription(string topicname)
{
var client = new ManagementClient(sbconnection);
bool topicExists = await client.TopicExistsAsync(topicname).ConfigureAwait(false);
string[] subscriptionarray = new string[] { "highprioritysubscription", "lowprioritysubscription" };
if (topicExists)
{
foreach (var item in subscriptionarray)
{
if (item == "highprioritysubscription")
{
SubscriptionDescription subName = new SubscriptionDescription(topicname, item);
subName.Status = 0;
RuleDescription subscriptionRule = new RuleDescription();
subscriptionRule.Filter = new SqlFilter("Priority >= 10");
var result01 = await client.CreateSubscriptionAsync(subName, subscriptionRule).ConfigureAwait(false);
}
else
{
SubscriptionDescription subName = new SubscriptionDescription(topicname, item);
subName.Status = 0;
RuleDescription subscriptionRule = new RuleDescription();
subscriptionRule.Filter = new SqlFilter("Priority < 10");
var result01 = await client.CreateSubscriptionAsync(subName, subscriptionRule).ConfigureAwait(false);
}
}
}
}
When publishing messages to a Service Bus topic, the client application has to set the custom-priority property so that the Service Bus topic can publish messages to separate virtual queues, and each client subscribing to the topic can be associated with either priority subscription:
public async Task SendMessageToTopicAsync()
{
TopicClient priorityClient = new TopicClient(sbconnection, "sample01");
for (int i = 0; i < 15; i++)
{
string messageBody = $"priorityTopic Message {i}";
var message = new Message(Encoding.UTF8.GetBytes(messageBody));
message.UserProperties.Add("Priority", i);
await priorityClient.SendAsync(message);
}
}
Based on the custom-priority property of the Service Bus topic, any subscribing application can subscribe to a message from the topic subscription. In this example, we have logic apps that subscribe to messages from the high-priority sample01 message topic. This technique of using the Service Bus topic and multiple subscription filters is advisable when you have a large set of Service Bus topic listeners.
In this section, we have covered how you can send and receive messages from virtual topic queues based on the custom property set for the message. In the next section, we will cover how you can leverage API Management as a message publisher to Azure Service Bus:
