• Azure.Messaging.ServiceBus
• Azure.Storage.Blobs
• Microsoft.Extensions.Configuration.Binder
• Microsoft.Extensions.Http
• System.Configuration.ConfigurationManager

Bu noktada, event’leri basit bir şekilde publish edip consume edebileceğimiz bir interface tanımlıyoruz. Bu service bus library’si içerisinde conditional bir şekilde claim-check pattern’ını implemente edeceğimiz için, Azure Service Bus client library’sinin kullanmakta olduğu “ServiceBusReceivedMessage” class’ını encapsulate ederek kendi “ServiceBusReceivedMessageWrapper” class’ımızı oluşturuyoruz. Bu tasarım, ilgili event’in payload’unun boyutuna göre, payload’un tutulacağı “Data” property’sini ya Azure Storage Account üzerinden ya da doğrudan Azure Service Bus üzerinden elde ederek hazırlayabilmemizi mümkün kılıyor olacak. Böylece, bu library’yi kullanan servisler, altyapıda gerçekleşen işlemlerden bağımsız bir şekilde event’lerini kusursuz bir şekilde consume etmeye devam ediyor olabilecekler.

Ayrıca, Azure Service Bus client library’si, consume edilen event’in tamamlanması işlemini gerçekleştirebilmek için kendi message tipi olan “ServiceBusReceivedMessage” objesine ihtiyaç duymaktadır. Bu nedenle, “ServiceBusReceivedMessage” objesini de wrapper içerisinde encapsulate ediyor ve tamamlanma işleminin de consumer’lar tarafında tutarlı bir şekilde gerçekleştirilebilmesi için “CompleteAsync” metodu ile dışarıya açıyor olacağız.

Şimdi yine aynı class içerisinde “IServiceBus” interface’inin implementasyonunu öncelikle “Publish” method’undan başlayarak parça parça ele alalım.

public class AzureServiceBus : IServiceBus
{
    private readonly ServiceBusClient _serviceBusClient;
    private readonly BlobContainerClient? _blobContainerClient;
    private readonly IHttpClientFactory _httpClientFactory;
    private readonly ConcurrentDictionary<string, ServiceBusSender> _senderPool;
    private readonly ConcurrentDictionary<string, ServiceBusReceiver> _receiverPool;
    private readonly int _maxEventPayloadSizeLimitInKB;
    private readonly int _claimCheckTokenExpirationInHours;
    private const string IsClaimCheckProperty = "IsClaimCheck";
    private const string ClaimCheckBlobUriProperty = "ClaimCheckBlobUri";
    private readonly bool _enableClaimCheck;
    private readonly string? _defaultQueueName;

    public AzureServiceBus(IConfiguration configuration, IHttpClientFactory httpClientFactory)
    {
        var serviceBusConnectionString = configuration.GetValue<string>("ServiceBus:ConnectionString") ?? throw new KeyNotFoundException("ServiceBus:ConnectionString is not found.");
        _defaultQueueName = configuration.GetValue<string>("ServiceBus:DefaultQueueName");
        _serviceBusClient = new ServiceBusClient(serviceBusConnectionString);

        _httpClientFactory = httpClientFactory;

        _senderPool = new ConcurrentDictionary<string, ServiceBusSender>();
        _receiverPool = new ConcurrentDictionary<string, ServiceBusReceiver>();


        _enableClaimCheck = configuration.GetValue<bool>("ServiceBus:ClaimCheck:EnableClaimCheck");

        if(_enableClaimCheck)
        {
            var blobStorageConnectionString = configuration.GetValue<string>("ServiceBus:ClaimCheck:BlobStorage:ConnectionString") ?? throw new KeyNotFoundException("ServiceBus:BlobStorage:ConnectionString is not found.");

            var claimCheckContainerName = configuration.GetValue<string>("ServiceBus:ClaimCheck:BlobStorage:ClaimCheckContainerName") ?? throw new KeyNotFoundException("ServiceBus:BlobStorage:ClaimCheckContainerName is not found.");

            _maxEventPayloadSizeLimitInKB = configuration.GetValue<int?>("ServiceBus:ClaimCheck:MaxEventPayloadSizeLimitInKB") ?? throw new KeyNotFoundException("ServiceBus:MaxEventPayloadSizeLimitInKB is not found.");

            _claimCheckTokenExpirationInHours = configuration.GetValue<int?>("ServiceBus:ClaimCheck:BlobStorage:ClaimCheckTokenExpirationInHours") ?? throw new KeyNotFoundException("ServiceBus:BlobStorage:ClaimCheckTokenExpirationInHours is not found.");

            _blobContainerClient = new BlobContainerClient(blobStorageConnectionString, claimCheckContainerName);
        }
    }

    public async Task Publish<T>(T @event, string? queueName = null, CancellationToken cancellationToken = default)
    {
        if(string.IsNullOrEmpty(queueName))
        {
            queueName = _defaultQueueName!;
        }

        var sender = _senderPool.GetOrAdd(queueName, _serviceBusClient.CreateSender);

        var serializedEvent = JsonSerializer.SerializeToUtf8Bytes(@event);

        ServiceBusMessage message;
        if (_enableClaimCheck && IsEventPayloadSizeExceedsLimit(serializedEvent))
        {
            var blobUri = await UploadPayloadToBlobAsync(serializedEvent, queueName, @event!.GetType().Name);

            message = new ServiceBusMessage
            {
                ApplicationProperties =
                {
                    [IsClaimCheckProperty] = true,
                    [ClaimCheckBlobUriProperty] = blobUri
                }
            };
        }
        else
        {
            message = new ServiceBusMessage(serializedEvent);
        }

        await sender.SendMessageAsync(message, cancellationToken);
    }

    private bool IsEventPayloadSizeExceedsLimit(byte[] serializedEvent)
    {
        var eventSize = serializedEvent.Length;
        Console.WriteLine(eventSize);
        return eventSize > _maxEventPayloadSizeLimitInKB * 1024;
    }

    private async Task<string> UploadPayloadToBlobAsync(byte[] serializedEvent, string queueName, string eventType)
    {
        var blobName = $"{queueName}/{eventType}/{Guid.NewGuid()}.json";

        await _blobContainerClient!.CreateIfNotExistsAsync();
        var blobClient = _blobContainerClient!.GetBlobClient(blobName);

        using var stream = new MemoryStream(serializedEvent);
        await blobClient.UploadAsync(stream, overwrite: true);

        var sasUri = blobClient.GenerateSasUri(Azure.Storage.Sas.BlobSasPermissions.Read, DateTimeOffset.UtcNow.AddHours(_claimCheckTokenExpirationInHours));

        return sasUri.ToString();
    }
}

Öncelikle, ihtiyaç duyduğumuz gerekli tüm konfigürasyon ve gerekli tüm servis injection işlemlerini gerçekleştiriyoruz. Claim-check kullanımıyla ilgili kararı consumer’lara bırakabilmek için ise, bu özelliğin kontrolünü “ServiceBus:ClaimCheck:EnableClaimCheck” parametresine bağlıyoruz. Ayrıca “ServiceBusSender” ve “ServiceBusReceiver” objelerini daha verimli yönetebilmek adına bir pooling mekanizması tanımlıyoruz.

“Publish” method’u ise asıl claim-check mekanizmasını conditional olarak implemente ettiğimiz ana nokta. Bu noktada, ilgili event ilk olarak serialize edilmektedir ve daha sonra, “IsEventPayloadSizeExceedsLimit” bloğu içerisinde conditional olarak claim-check işlemi gerçekleştirilmektedir. Eğer ilgili event’in boyutu belirlenmiş sınırın üzerinde ise, claim-check işlemi otomatik olarak Azure Blob Storage üzerinden gerçekleştirilecektir.

Bu noktada sırasıyla:

• “UploadPayloadToBlobAsync” methud’u içerisinde ilgili event payload’unu Azure Blob Storage üzerine upload etme işlemini ele alıyoruz.
• Ardından ilgili payload’un Azure Blob Storage üzerinden tekrar elde edilebilmesi için ise bir SAS URI oluşturuyoruz.
• Daha sonra ise oluşturmuş olduğumuz bu SAS URI bilgisini, ilgili event’in metadata bilgileri içerisine “ApplicationProperties” dahil ediyoruz.

Şimdi ise “Receive” method’u ile devam edelim.

public class AzureServiceBus : IServiceBus
{
    // other codes...

    public async Task<ServiceBusMessageReceivedMessageWrapper<T>> Receive<T>(string? queueName = null, CancellationToken cancellationToken = default)
    {
        if (string.IsNullOrEmpty(queueName))
        {
            queueName = _defaultQueueName!;
        }

        var receiver = _receiverPool.GetOrAdd(queueName, _serviceBusClient.CreateReceiver);

        var message = await receiver.ReceiveMessageAsync(cancellationToken: cancellationToken);

        ServiceBusReceivedMessageWrapper<T> wrappedMessage;

        if (message.ApplicationProperties.TryGetValue(IsClaimCheckProperty, out var isClaimCheck) && (bool)isClaimCheck)
        {
            var blobUri = message.ApplicationProperties[ClaimCheckBlobUriProperty] as string
                  ?? throw new InvalidOperationException($"The claim-check value is not found. Message CorrelationId: {message.CorrelationId}");

            byte[] payload = await DownloadPayloadFromBlobAsync((string)blobUri);

            var actualEvent = JsonSerializer.Deserialize<T>(payload);

            wrappedMessage = new ServiceBusReceivedMessageWrapper<T>(message, receiver, actualEvent!);
        }
        else
        {
            wrappedMessage = new ServiceBusReceivedMessageWrapper<T>(message, receiver, message.Body.ToObjectFromJson()!);
        }

        return wrappedMessage;
    }

    private async Task<byte[]> DownloadPayloadFromBlobAsync(string blobUri)
    {
        var httpClient = _httpClientFactory.CreateClient();
        using var response = await httpClient.GetAsync(blobUri);

        if (response.IsSuccessStatusCode)
        {
            return await response.Content.ReadAsByteArrayAsync();
        }

        throw new Exception($"Failed to download the claim-check payload from {blobUri}. Status Code: {response.StatusCode}");
    }
}

Bu noktada, ilgili event’i consume ettikten sonra metadata bilgileri içerisinde claim-check token’ının yer alıp almadığını kontrol ediyoruz. Eğer claim-check token’ı mevcutsa, metadata bilgileri içerisinden Blob SAS URI bilgisini elde ediyor ve “DownloadPayloadFromBlobAsync” method’unu kullanarak ilgili event payload’unu elde ediyoruz. Claim-check token’ı bulunmuyor ise, ilgili event’i consume ettiğimiz gibi wrapper aracılığıyla ilgili consumer’a iletiyoruz. Böylece, daha önce de bahsettiğimiz gibi, consumer’lar altyapıda gerçekleşen claim-check işlemlerinden bağımsız bir şekilde event’leri consume etmeye devam edecektir.

Şimdi, hızlıca bir test işlemi gerçekleştirebilmek için “OrderCreatedEvent” ‘ini publish edecek olan publisher’ı implemente edelim. İlk olarak, “TodoEcom.Contracts” adında bir class library oluşturalım ve içerisinde publisher ve consumer arasında paylaşacak olduğumuz “OrderCreatedEvent” ini aşağıdaki gibi tanımlayalım.

namespace TodoEcom.Contracts.Events;

public class OrderCreatedEvent
{
    public int OrderId { get; set; }
    public int CustomerId { get; set; }
    public required string PaymentMethod { get; set; }
    public DateTime CreatedAt { get; set; }
    public required IReadOnlyList<ProductDTO> Products { get; set; }

    // other attributes such as address, applied discounts, shipping details, transaction details and etc.
}

public class ProductDTO
{
    public int ProductId { get; set; }
    public required string Name { get; set; }
    public int Quantity { get; set; }
    public decimal Price { get; set; }
}

Bu noktada, örneği basit tutmak adına çok fazla attribute eklemedim. Ancak ana fikri iletebildiğimi düşünüyorum.

Şimdi ise “TodoEcom.OrderService.API” adında bir .NET 9 webapi projesi oluşturalım ve “TodoEcom.Contracts” ile “TodoEcom.ServiceBus” library’lerini referans olarak dahil edelim. Ardından “Program.cs” class’ını aşağıdaki gibi güncelleyelim.

using TodoEcom.Contracts.Events;
using TodoEcom.ServiceBus;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddOpenApi();
builder.Services.AddHttpClient();
builder.Services.AddSingleton<IServiceBus, AzureServiceBus>();

var app = builder.Build();

if (app.Environment.IsDevelopment())
{
    app.MapOpenApi();
}

app.UseHttpsRedirection();

app.MapPost("/orders", async (IServiceBus serviceBus) =>
{
    // Perform order operations...

    var orderCreatedEvent = new OrderCreatedEvent
    {
        OrderId = 1,
        CustomerId = 123,
        PaymentMethod = "CreditCard",
        CreatedAt = DateTime.UtcNow,
        Products = Enumerable.Range(1,20).Select(i => new ProductDTO
        {
            ProductId = i,
            Name = $"My product {i}",
            Quantity = 1,
            Price = i,
        }).ToList()
    };

    await serviceBus.Publish(orderCreatedEvent);

})
.WithName("CreateAnOrder");

app.Run();

Bu noktada, gerekli client’ların registration işlemlerini gerçekleştirdikten sonra implemente etmiş olduğumuz service bus library’sini kullanarak örnek bir “OrderCreatedEvent” ‘i publish ediyoruz. Claim-check mekanizmasını test edebilmek için ise, publish edecek olduğumuz bu event’in belirleyecek olduğumuz payload limitini aşabilmesi adına içerisine birkaç ürün bilgisi ekliyoruz.

Şimdi “appsettings.json” dosyasını ise aşağıdaki gibi güncelleyelim.

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "ServiceBus": {
    "ConnectionString": "Endpoint=sb://mypocsbus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=YOUR_SHARED_ACCESS_KEY",
    "DefaultQueueName": "orders",
    "ClaimCheck": {
      "EnableClaimCheck": true,
      "MaxEventPayloadSizeLimitInKB": 1,
      "BlobStorage": {
        "ConnectionString": "DefaultEndpointsProtocol=https;AccountName=mypocstorageac;AccountKey=YOUR_ACCOUNT_KEY;EndpointSuffix=core.windows.net",
        "ClaimCheckContainerName": "claim-check",
        "ClaimCheckTokenExpirationInHours": 1
      }
    }
  }
}

Bu noktada, “EnableClaimCheck” parametresi ile claim-check özelliğini kullanacağımızı belirtiyor, “MaxEventPayloadSizeLimitInKB” parametresi ile ise Azure Service Bus üzerine publish etmek istediğimiz event boyutunu 1 KB ile sınırlıyoruz.

Şimdi, “TodoEcom.BillingService.Consumer” adında bir .NET 9 worker projesi oluşturalım ve “TodoEcom.Contracts” ile “TodoEcom.ServiceBus” library’lerini referans olarak ekleyelim. Ardından, “OrderCreatedEvent” ‘inin consume işlemini gerçekleştirelim.

using TodoEcom.Contracts.Events;
using TodoEcom.ServiceBus;

namespace TodoEcom.BillingService.Consumer;

public class Worker : BackgroundService
{
    private readonly ILogger<Worker> _logger;
    private readonly IServiceBus _serviceBus;
    private const string OrdersQueueName = "orders";

    public Worker(ILogger<Worker> logger, IServiceBus serviceBus)
    {
        _logger = logger;
        _serviceBus = serviceBus;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            if (_logger.IsEnabled(LogLevel.Information))
            {
                _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
            }

            var @event = await _serviceBus.Receive<OrderCreatedEvent>(OrdersQueueName, stoppingToken);

            Console.WriteLine($"Event received. OrderId: {@event.Data.OrderId} ");

            // some operations...

            await @event.CompleteAsync();
        }
    }
}

“appsettings.json” dosyasını da aşağıdaki gibi güncelleyelim.

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  },
  "ServiceBus": {
    "ConnectionString": "Endpoint=sb://mypocsbus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=YOUR_SHARED_ACCESS_KEY"
  }
}

Artık test etmeye hazırız.

Test Edelim

İlk olarak “TodoEcom.OrderService.API” projesini çalıştıralım ve “/orders” endpoint’ini trigger edelim.

curl -X POST http://localhost:5065/orders

Ardından, Azure Portal üzerinden “Service Bus Explorer” ‘ı kullanarak, “orders” queue’suna publish etmiş olduğumuz message’ın “Custom Properties” bölümüne bir bakalım.

Bu noktada, publish ettiğimiz event payload’unun Azure Blob Storage ‘a yüklenip, ilgili Blob SAS URI bilgisinin “ClaimCheckBlobUri” property’sine eklendiğini görebiliriz.

Ardından “TodoEcom.BillingService.Consumer” projesini çalıştıralım ve ilgili event’i consume edelim.

Terminal log’larından da görebileceğimiz gibi, publish etmiş olduğumuz event, geliştirmiş olduğumuz service bus library’si tarafından consume edildikten sonra payload bilgisi başarıyla ilgili storage account üzerinden alınmış ve “TodoEcom.BillingService.Consumer” projesine kesintisiz bir şekilde consume edebilmesi için iletilmiştir.

Bu makale kapsamında, conditional claim-check mekanizmasını büyük payload’ları yönetme süreçlerinde basit bir şekilde nasıl kullanabileceğimizi ele almaya çalıştık. Implemente etmiş olduğumuz service bus library’si sayesinde, ilgili consumer’lar altyapı işlemlerinden bağımsız olarak event’lerini kesintisiz bir şekilde publish ve consume etmeye devam ederken, sistemin ölçeklenebilirliğinin ve verimliliğinin minimum düzeyde etkilenebilmesini sağlamaya çalıştık.

Örnek kodlara buradan ulaşabilirsiniz: https://github.com/GokGokalp/claim-check-sample

Referanslar

https://learn.microsoft.com/en-us/azure/architecture/patterns/claim-check
https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-get-started-with-queues