Azure Service Bus
Azure Service Bus
Legacy was MSMQ. Azure Service Bus is better because: 1) No Memory Locks, 2) Dead-Lettering, 3) Scaling.
Namespaces in C#:
Azure.Messaging.ServiceBus (latest)
Microsoft.Azure.ServiceBus (legacy - as of 30 Sep 2026)
Basically you will be doing the typical pub/sub that is decoupled. Events should be immutable (so "record" type).
Advantages: Loose Coupling, High Scalability, Real-time Responsiveness
Challenges: Eventual Consistency, Debugging Is Hard, Payload Overhead
Azure Service Bus is primarily a Broker topology (rather than a mediator).
Steps:
1) Event:
public record OrderPlacedEvent(Guid OrderId, string CustomerEmail, decimal TotalAmount);
2) Producer (Publishes to Azure Topic):
using System.Text.Json;
using Azure.Messaging.ServiceBus;
public class AzureEventPublisher
{
private readonly ServiceBusSender _sender;
public AzureEventPublisher(ServiceBusClient client, string topicName)
{
_sender = client.CreateSender(topicName);
}
public async Task PublishAsync<TEvent>(TEvent @event)
{
string jsonBody = JsonSerializer.Serialize(@event);
var message = new ServiceBusMessage(jsonBody) { ContentType = "application/json", Subject = typeof(TEvent).Name };
await _sender.SendMessageAsync(message);
Console.WriteLine($"[Publisher] Successfully broadcasted {typeof(TEvent).Name} to Azure Service Bus.");
}
}
3) Consumer Handler (Listen to Azure Subscription):
using System.Text.Json;
using Azure.Messaging.ServiceBus;
public class AzureSubscriptionListener
{
private readonly ServiceBusProcessor _processor;
public AzureSubscriptionListener(ServiceBusClient client, string topicName, string subscriptionName)
{
var options = new ServiceBusProcessorOptions { AutoCompleteMessages = false, MaxConcurrentCalls = 2 };
_processor = client.CreateProcessor(topicName, subscriptionName, options);
}
public async Task StartListeningAsync()
{
_processor.ProcessMessageAsync += MessageHandler;
_processor.ProcessErrorAsync += ErrorHandler;
await _processor.StartProcessingAsync();
}
private async Task MessageHandler(ProcessMessageEventArgs args)
{
// 1. Extract payload
string rawJson = args.Message.Body.ToString();
// 2. Deserialize back into our local C# structured record
var orderEvent = JsonSerializer.Deserialize<OrderPlacedEvent>(rawJson);
if (orderEvent != null)
{
// 3. Execute the decoupled handler BL
Console.WriteLine($"[Consumer Handler] Received Order {orderEvent.OrderId}. Sending confirmation email to {orderEvent.CustomerEmail}...");
await Task.Delay(500);
}
// 4. Acknowledge and settle the message. It is now safely removed from Azure Service Bus.
await args.CompleteMessageAsync(args.Message);
}
private Task ErrorHandler(ProcessErrorEventArgs args)
{
// Errors (network dropouts, authorization expiration, etc.) end up here
Console.WriteLine($"[Error Handler] Transport error occurred: {args.Exception.Message}");
return Task.CompletedTask;
}
public async Task StopListeningAsync()
{
await _processor.StopProcessingAsync();
await _processor.DisposeAsync();
}
}
4) Flow:
using Azure.Identity;
using Azure.Messaging.ServiceBus;
class Program
{
static async Task Main()
{
string fullyQualifiedNamespace = "your-servicebus-namespace.servicebus.windows.net";
string topicName = "order-events-topic";
string emailSubscription = "send-email-on-order";
// 1. Initialize the client with Entra ID/Managed Identity
var credential = new DefaultAzureCredential();
await using var client = new ServiceBusClient(fullyQualifiedNamespace, credential);
// 2. Boot up the background listener (Consumer Handler)
var emailListener = new AzureSubscriptionListener(client, topicName, emailSubscription);
await emailListener.StartListeningAsync();
Console.WriteLine("Azure Event Listener active and polling for messages...");
// 3. Boot up the event sender (Producer)
var publisher = new AzureEventPublisher(client, topicName);
// 4. Simulate a user event triggering the publish pipeline
var mockEvent = new OrderPlacedEvent(Guid.NewGuid(), "cloudbuyer@example.com", 149.99m);
await publisher.PublishAsync(mockEvent);
// Keep console app running long enough to watch the background listener handle the roundtrip
await Task.Delay(3000);
// Clean up
await emailListener.StopListeningAsync();
}
}
===============================
Azure Durable Functions (Orchestrations): If your business flow dictates that a system must first charge a credit card, then check inventory, and finally generate a shipping label—rolling back previous steps if one fails—you layer Azure Durable Functions on top of your queues. The Durable Function acts as the central Mediator (Orchestrator), using Azure Service Bus queues as communication channels to send explicit commands to individual workers.
1) Orders:
public record OrderRequest(Guid OrderId, string CustomerId, decimal Amount, string ItemSku, int Quantity);
public record OrderResult(bool Success, string Details);
2) Orchestrator:
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
public class OrderSagaOrchestrator
{
[Function(nameof(RunOrderWorkflow))]
public async Task<OrderResult> RunOrderWorkflow(
[OrchestrationTrigger] TaskOrchestrationContext context)
{
var log = context.CreateReplaySafeLogger<OrderSagaOrchestrator>();
var order = context.GetInput<OrderRequest>();
bool creditCardCharged = false;
bool inventoryReserved = false;
try
{
// Step 1: Charge the Credit Card
log.LogInformation($"[Orchestrator] Step 1: Charging card for order {order.OrderId}");
await context.CallActivityAsync(nameof(ChargeCreditCardActivity), order);
creditCardCharged = true;
// Step 2: Check and Reserve Inventory
log.LogInformation($"[Orchestrator] Step 2: Checking inventory for SKU {order.ItemSku}");
await context.CallActivityAsync(nameof(ReserveInventoryActivity), order);
inventoryReserved = true;
// Step 3: Generate the Shipping Label
log.LogInformation($"[Orchestrator] Step 3: Generating shipping label");
await context.CallActivityAsync(nameof(GenerateShippingLabelActivity), order);
log.LogInformation($"[Orchestrator] Workflow completed successfully for order {order.OrderId}");
return new OrderResult(true, "Order completed and shipped successfully.");
}
catch (Exception ex)
{
log.LogError($"[Orchestrator] Critical failure occurred: {ex.Message}. Starting compensation rollback...");
await RollbackTransactionAsync(context, order, creditCardCharged, inventoryReserved, log);
return new OrderResult(false, $"Order failed and was safely rolled back. Reason: {ex.Message}");
}
}
private async Task RollbackTransactionAsync(
TaskOrchestrationContext context,
OrderRequest order,
bool creditCardCharged,
bool inventoryReserved,
ILogger log)
{
if (inventoryReserved)
{
log.LogWarning($"[Rollback] Releasing reserved stock for SKU {order.ItemSku}");
await context.CallActivityAsync(nameof(RefundInventoryActivity), order);
}
if (creditCardCharged)
{
log.LogWarning($"[Rollback] Refunding payment of {order.Amount:C} to customer {order.CustomerId}");
await context.CallActivityAsync(nameof(RefundCreditCardActivity), order);
}
log.LogInformation($"[Rollback] Compensation complete for order {order.OrderId}. System is back in a consistent state.");
}
}
3) Workers:
public class OrderActivities
{
// --- STEP 1: PAYMENT WORKERS ---
[Function(nameof(ChargeCreditCardActivity))]
public Task ChargeCreditCardActivity([ActivityTrigger] OrderRequest order, FunctionContext executionContext)
{
// Add actual credit card payment gateway integration code here
Console.WriteLine($"[Payment API] Successfully charged {order.Amount:C} for Order {order.OrderId}.");
return Task.CompletedTask;
}
[Function(nameof(RefundCreditCardActivity))]
public Task RefundCreditCardActivity([ActivityTrigger] OrderRequest order, FunctionContext executionContext)
{
// Reverse payment gateway transaction
Console.WriteLine($"[Payment API] Refund executed for Order {order.OrderId}.");
return Task.CompletedTask;
}
// --- STEP 2: INVENTORY WORKERS ---
[Function(nameof(ReserveInventoryActivity))]
public Task ReserveInventoryActivity([ActivityTrigger] OrderRequest order, FunctionContext executionContext)
{
// Simulate a business exception logic check
if (order.Quantity > 50)
{
throw new InvalidOperationException($"Out of Stock! Cannot fulfill requested quantity of {order.Quantity}.");
}
Console.WriteLine($"[Inventory DB] Stock allocated for SKU {order.ItemSku} (Qty: {order.Quantity}).");
return Task.CompletedTask;
}
[Function(nameof(RefundInventoryActivity))]
public Task RefundInventoryActivity([ActivityTrigger] OrderRequest order, FunctionContext executionContext)
{
// Re-increment stock count back into your database system
Console.WriteLine($"[Inventory DB] Restocked {order.Quantity} units of SKU {order.ItemSku} back to shelves.");
return Task.CompletedTask;
}
// --- STEP 3: SHIPPING WORKER ---
[Function(nameof(GenerateShippingLabelActivity))]
public Task GenerateShippingLabelActivity([ActivityTrigger] OrderRequest order, FunctionContext executionContext)
{
Console.WriteLine($"[Shipping API] Courier dispatch confirmed. Label printed for Order {order.OrderId}.");
return Task.CompletedTask;
}
}
4) Workers:
using System.Net;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.DurableTask.Client;
public class OrderClientTrigger
{
[Function(nameof(HttpStartOrderSaga))]
public async Task<HttpResponseData> HttpStartOrderSaga(
[HttpTrigger(AuthorizationLevel.Function, "post", Route = "checkout")] HttpRequestData req,
[DurableClient] DurableTaskClient client,
FunctionContext executionContext)
{
// 1. Read input request payload
var orderData = await req.ReadFromJsonAsync<OrderRequest>();
if (orderData == null)
{
return req.CreateResponse(HttpStatusCode.BadRequest);
}
// 2. Schedule the orchestration to execute in the background asynchronously
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
nameof(OrderSagaOrchestrator.RunOrderWorkflow),
orderData
);
// 3. Return a tracking management response containing status endpoints instantly
var response = req.CreateResponse(HttpStatusCode.Accepted);
await response.WriteAsJsonAsync(new { OrchestrationInstanceId = instanceId });
return response;
}
}
Comments
Post a Comment