Introduction
In our earlier article, I coated an summary of the service bus, particularly on queue & subjects. Lastly, we noticed how the queue works from the coding perspective.
Configuring Matter in Azure Portal
Let’s bounce straight into learn how to create a service bus-topic from the Azure portal.
I am preferring to stick with the default settings whereas creating the subject i.e. partitioning and duplicate detection are set to disable.
Right here, I am creating two subscriptions, one with Order subscription and one other one with Product subscription. In each subscriptions, we’ve solicited service bus to have the utmost deliveries of messages to five, nevertheless, the remaining settings are set to default. In future articles, we’ll cowl periods, useless lettering, and many others.
I’ve created two shared entry insurance policies, one for ship and one other one for listening (i.e. we’ll use ship coverage within the writer, whereas the listening coverage for customers (merchandise and orders customers).
Coding
I’ve created four tasks for this text:
- ASP.NET Core net api-> Producer
- 2 Employee Service projects-> 1 for merchandise shopper and one other one for orders shopper.
- .NET core class library-> Widespread lessons
Add a Microsoft.Azure.ServiceBus NuGet bundle within the producer and shopper (merchandise and orders) tasks
Firstly, let’s sort out the Producer performance:
- public interface IMessagePublisher {
- Job PublisherAsync < T > (T request);
- }
- public class MessagePublisher: IMessagePublisher {
- non-public readonly ITopicClient topicClient;
- public MessagePublisher(ITopicClient topicClient) {
- this.topicClient = topicClient;
- }
- public async Job PublisherAsync < T > (T request) {
- var message = new Message {
- MessageId = Guid.NewGuid().ToString(),
- Physique = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(request))
- };
- message.UserProperties.Add(“MessageType”, typeof(T).Identify);
- await topicClient.SendAsync(message);
- }
- }
Message Writer class is just about much like what we’ve carried out within the earlier article. As a substitute of IQueueClient, we’ve to make use of ITopicClient for subject; furthermore, I’ve added MessageId and customized property(MessageType) to the message header. Later, we’ll talk about about why I added customized property intimately.
- [Route(“api/[controller]”)]
- [ApiController]
- public class TopicController: ControllerBase {
- non-public readonly IMessagePublisher messagePublisher;
- public TopicController(IMessagePublisher messagePublisher) {
- this.messagePublisher = messagePublisher;
- }
- [HttpPost(template: “product”)]
- public async Job SendToProduct([FromBody] Product product) {
- await messagePublisher.PublisherAsync(product);
- }
- [HttpPost(template: “order”)]
- public async Job SentToOrder([FromBody] Order order) {
- await messagePublisher.PublisherAsync(order);
- }
- }
I’ve created a reasonably simple controller with an HTTP verb as submit for each product and order.
- public void ConfigureServices(IServiceCollection companies) {
- companies.AddControllers();
- companies.AddSingleton < ITopicClient > (serviceProvider => new TopicClient(connectionString: Configuration.GetValue < string > (“servicebus:connectionstring”), entityPath: Configuration.GetValue < string > (“serviceBus:topicname”)));
- companies.AddSingleton < IMessagePublisher, MessagePublisher > ();
- }
Within the startup class, I’ve created a dependency injection for ITopicClient and IMessagePublisher.
- {
- “servicebus”: {
- “connectionstring”: “<ConnectionString Right here>”,
- “topicname”: “<Matter identify right here>”
- },
- “Logging”: {
- “LogLevel”: {
- “Default”: “Info”,
- “Microsoft”: “Warning”,
- “Microsoft.Internet hosting.Lifetime”: “Info”
- }
- },
- “AllowedHosts”: “*”
- }
Listed here are the area lessons outlined in a standard challenge.
- public class Order {
- public int Id {
- get;
- set;
- }
- public string Identify {
- get;
- set;
- }
- public int Amount {
- get;
- set;
- }
- }
- public class Product {
- public int Id {
- get;
- set;
- }
- public string Identify {
- get;
- set;
- }
- public int Worth {
- get;
- set;
- }
- public string ProductStatus {
- get;
- set;
- }
- }
Now run the writer with both merchandise or orders. In my case, I am operating with merchandise:
Now confirm the message rely within the subscription:
The great half is that we’re getting the message rely as 1 however wait we’ve the message rely as 1 for orders too. Bizarre, we’re sending the message for merchandise however we’ve obtained a message for orders as properly. Any guess as to why?
The reply is as a result of we have no filters for the subject.
Let’s perceive what a filter is in subject. Subscribers need to outline which message they need from the subject is known as a filter. Every of the newly created subject subscriptions has an preliminary default subscription rule. If in case, you do not explicitly specify a filter situation for the rule, the utilized filter is 1=1 that permits all messages to be chosen into the subscription.
There are 2 varieties of filters in subjects:
- SQL Filters- SQL filters maintain SQL like conditional expressions in opposition to system and consumer properties(customized properties). Right here is the reply, why we used customized property within the message.
- Correlational Filters- Correlation filters are used to match in opposition to a number of message techniques or consumer properties.
In these articles, we’ll be targeted on each filters.
By default, the default filter is utilized i.e. 1=1 or it is going to settle for all of the incoming messages.
Let’s add a brand new filter:
For merchandise, I’ve used the SQL filter. With a view to obtain a message from the subject, I’ve set MessageType=’Product’
For orders, I’ve used a Correlation filter. I’ve specified the customized properties to obtain a message from the subject.
Now run the postman and see the end result. Now I am operating for merchandise:
Right here we go, the message has been obtained for product subscription. The writer appears to be working effective.
Now, let’s transfer our deal with each the customers. First, let’s start with merchandise shopper.
- public static IHostBuilder CreateHostBuilder(string[] args) =>
- Host.CreateDefaultBuilder(args)
- .ConfigureServices((hostContext, companies) =>
- {
- companies.AddHostedService<Employee>();
- companies.AddSingleton<ISubscriptionClient>(serviceProvider => new SubscriptionClient(
- connectionString: “<ConnectionString Right here”,
- topicPath: “<Matter Identify right here>”, subscriptionName: “ProductSubscription”));
- });
We now have created dependency injection for ISubscriptionClient by passing connection string, subject identify and subscription identify.
Word
It’s a must to use hear key of the shared entry coverage for the connection string.
- public class Employee: BackgroundService {
- non-public readonly ILogger < Employee > _logger;
- non-public readonly ISubscriptionClient subscriptionClient;
- public Employee(ILogger < Employee > logger, ISubscriptionClient subscriptionClient) {
- _logger = logger;
- this.subscriptionClient = subscriptionClient;
- }
- protected override async Job ExecuteAsync(CancellationToken stoppingToken) {
- subscriptionClient.RegisterMessageHandler((message, token) => {
- _logger.LogInformation($ “message id:{message.MessageId}”);
- _logger.LogInformation($ “message physique:{Encoding.UTF8.GetString( message.Physique)}”);
- var product = JsonConvert.DeserializeObject < Widespread.Product > (Encoding.UTF8.GetString(message.Physique));
- return subscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
- }, new MessageHandlerOptions(ex => {
- _logger.LogError(ex.Exception, ex.Exception.Message);
- return Job.FromException(ex.Exception);
- }) {
- AutoComplete = false,
- MaxConcurrentCalls = 1
- });
- }
You might want to register the message handler with a view to obtain messages from the subject. Within the above snippet, we’re changing the message physique from bytes to product object. Do not forget to specify AutoComplete=false different you’ll find yourself with an exception.
Equally, we’ve to create for Order subscription as properly. There are completely no change when it comes to code apart from specifying subscription identify in this system class.
- public static IHostBuilder CreateHostBuilder(string[] args) =>
- Host.CreateDefaultBuilder(args)
- .ConfigureServices((hostContext, companies) =>
- {
- companies.AddHostedService<Employee>();
- companies.AddSingleton<ISubscriptionClient>(serviceProvider => new SubscriptionClient(
- connectionString: “<ConnectionString Right here”,
- topicPath: “<Matter Identify right here>”, subscriptionName: “ProductSubscription”));
- });
Lastly, we’ve managed to place all of the code modifications in place. Run the appliance and confirm for each the merchandise and orders.
I hope you favored the article. In case you discover the article fascinating, then kindly like and share it.
