Azure Service Bus - Send & Receive Message - Topic in .Net Core

Azure Service Bus - Send & Receive Message - Topic in .Net Core

I have seen many examples to send and receive message of Azure Service Bus Topic in normal .Net framework, but I couldn't find correct code for .Net Core for the same to send and receive Azure Service Bus Topic.

Finally, I got solution and here is code for .Net Core C#.

Send Message

1. Create web api project with .Net core
2. Add Service Bus NuGet package, search for "Microsoft.Azure.ServiceBus"
3. Add NuGet package for "Newtonsoft.Json"
4. Create Test API controller
5. Write below code

using System;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc; 
using Microsoft.Azure.ServiceBus;
using Newtonsoft.Json; 
using System.Collections.Generic; 
using System.Text; 

namespace Demo.WebAPI.Controllers
{
    [Produces("application/json")]
    [Route("api/Test")]
    public class TestController : Controller
    {
        [HttpPost]
        public async Task<IActionResult> Post(dynamic model)
        {
            try
            {
                await Task.Run(() =>
                {
                    MessageFactory.SendMessagesAsync("TestTopic",
                           "testLable", 60, model);
                });
            }
            catch (Exception ex)
            {

                return Ok(ex.Message);
            }

            return Ok("Request submitted. ");
        }
        public static async Task SendMessagesAsync(string topicName, string lable, int minsTimeToLive, dynamic model, Dictionary<string, string> properties = null)
        {
            try
            {
                ITopicClient topicClient = new TopicClient(Config.ConnectionString, topicName);

                Message message = new Message(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(model)))
                {
                    ContentType = "application/json",
                    Label = "test",
                    TimeToLive = TimeSpan.FromMinutes(minsTimeToLive)
                };

                await topicClient.SendAsync(message);
            }
            catch (Exception exception)
            {

            }
        }
    } 
}

Test Web API 

1. Open Postman or restclient to test web api
2. URL :http://localhost/Demo.WebAPI/api/Test
2. Content-Type = application/json
3. Type: POST
4. Body as below,

 {
  "OrderDate":"02/22/2018",
  "CustomerId": 28,
  "Amount":35.23,
  "PaymentType":"Online",
  "Products":[
   {
    "ProductId":1,
    "Quantity":2,
    "Amount":10
   },
   {
    "ProductId":3,
    "Quantity":1,
    "Amount":15
   }
   ],
   "ShippingAddress":
   {
    "ShippingBy":"BlueDart",
    "CustomerName":"Test Customer",
    "Address":"Address 1",
  "Email":"test@test.com",
  "Phone":"1234567890"
   }
}

After posting request, you should get Status = 200 OK


Receive Message

1. Create .Net Core console application
2. Add reference using nuget for "Microsoft.Azure.ServiceBus" and "Newtonsoft.Json"
3. Write below code to receive azure service bus topic message

using System;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.ServiceBus;
using Newtonsoft.Json; 

namespace Demo.Service
{
    class Program
    {
        const string ServiceBusConnectionString = "Endpoint=sb://[NameSpace].servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=[SharedAccessKey]=";
        const string TopicName = "TestTopic";
        const string SubscriptionName = "TestSubscriber";
        static ISubscriptionClient subscriptionClient;

        static void Main(string[] args)
        {
            Console.WriteLine("Service started...");

            while (true)
            {
                MainAsync().GetAwaiter().GetResult();
            }

        }

        static async Task MainAsync()
        {
            subscriptionClient = new SubscriptionClient(ServiceBusConnectionString, TopicName, SubscriptionName);

            RegisterOnMessageHandlerAndReceiveMessages();

            await subscriptionClient.CloseAsync();
        }

        static void RegisterOnMessageHandlerAndReceiveMessages()
        {
            // Configure the message handler options in terms of exception handling, number of concurrent messages to deliver, etc.
            var messageHandlerOptions = new MessageHandlerOptions(ExceptionHandler)
            {
                // Maximum number of concurrent calls to the callback ProcessMessagesAsync(), set to 1 for simplicity.
                // Set it according to how many messages the application wants to process in parallel.
                MaxConcurrentCalls = 1,

                // Indicates whether MessagePump should automatically complete the messages after returning from User Callback.
                // False below indicates the Complete will be handled by the User Callback as in `ProcessMessagesAsync` below.
                AutoComplete = false
            };

            // Register the function that processes messages.
            subscriptionClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
        }

        static async Task ProcessMessagesAsync(Message message, CancellationToken token)
        {
            var a = new MemoryStream(message.Body);

            var body = Encoding.UTF8.GetString(message.Body);

            Order order = JsonConvert.DeserializeObject<Order>(body);

            // Process the message.
            Console.WriteLine($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");
            // Console.WriteLine($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Id = { order.OrderId}");
            // Complete the message so that it is not received again.
            // This can be done only if the subscriptionClient is created in ReceiveMode.PeekLock mode (which is the default).
            await subscriptionClient.CompleteAsync(message.SystemProperties.LockToken);

            // Note: Use the cancellationToken passed as necessary to determine if the subscriptionClient has already been closed.
            // If subscriptionClient has already been closed, you can choose to not call CompleteAsync() or AbandonAsync() etc.
            // to avoid unnecessary exceptions.
        }

        static Task ExceptionHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
        {
            Console.WriteLine($"Message handler encountered an exception {exceptionReceivedEventArgs.Exception}.");
            var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
            Console.WriteLine("Exception context for troubleshooting:");
            Console.WriteLine($"- Endpoint: {context.Endpoint}");
            Console.WriteLine($"- Entity Path: {context.EntityPath}");
            Console.WriteLine($"- Executing Action: {context.Action}");
            return Task.CompletedTask;
        }
    }
}

Add classes to do business process....

public class Order
{
 public Guid OrderId { get; set; }
 public int CustomerId { get; set; }
 public string InvoiceNo { get; set; }
 public DateTime OrderDate { get; set; }
 public bool Payment { get; set; }
 public string OrderStatus { get; set; }
 public string PaymentType { get; set; }
 public decimal Amount { get; set; }

 public List<OrderItems> Products { get; set; }
 public ShippingAddress ShippingAddress { get; set; }
}

public class OrderItems
{
 public int ProductId { get; set; }
 public int Quantity { get; set; }
 public decimal Amount { get; set; }
}

public class ShippingAddress
{
 public Int64 ShippingId { get; set; }
 public Guid OrderId { get; set; }
 public DateTime ShippingDate { get; set; }
 public string ShippingBy { get; set; }
 public string CustomerName { get; set; }
 public string Address { get; set; }
 public string Email { get; set; }
 public string Phone { get; set; }
 public string ShippingStatus { get; set; }

}

Run console application and you will receive azure service bus message.



Enjoy !!!

:)