diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 9161fc1..42f0f90 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -1,41 +1,40 @@ -name: .NET - -on: - push: - branches: [ master ] - pull_request: - branches: [ master ] - -jobs: - build: - - runs-on: ubuntu-latest - - services: - # Azurite emulates Azure Storage so the integration tests run without a - # live storage account. The official image binds to 0.0.0.0, so the - # mapped ports are reachable from the runner at 127.0.0.1. - azurite: - image: mcr.microsoft.com/azure-storage/azurite - ports: - - 10000:10000 - - 10001:10001 - - 10002:10002 - - steps: - - uses: actions/checkout@v3 - - name: Setup .NET - uses: actions/setup-dotnet@v3 - with: - dotnet-version: | - 8.0.x - - name: Restore dependencies - run: dotnet restore - - name: Build - run: dotnet build --no-restore - - name: Test - env: - # UseDevelopmentStorage=true targets the Azurite emulator above. - Storage__ConnectionString: "UseDevelopmentStorage=true" - Storage__QueueName: "integration-test-queue" - run: dotnet test --no-build --verbosity normal -p:ParallelizeTestCollections=false \ No newline at end of file +name: .NET + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + # Azurite emulates Azure Storage so the integration tests run without a + # live storage account. It is started with docker run rather than as a + # service container because it needs --skipApiVersionCheck: the Azure SDK + # sends an API version newer than Azurite knows about and Azurite rejects + # it with a 400 otherwise. Only the queue port (10001) is mapped. + - name: Start Azurite + run: | + docker run -d --name azurite -p 10001:10001 \ + mcr.microsoft.com/azure-storage/azurite \ + azurite --queueHost 0.0.0.0 --blobHost 0.0.0.0 --tableHost 0.0.0.0 --skipApiVersionCheck --silent + - name: Setup .NET + uses: actions/setup-dotnet@v4 + with: + dotnet-version: | + 8.0.x + - name: Restore dependencies + run: dotnet restore + - name: Build + run: dotnet build --no-restore + - name: Test + env: + # UseDevelopmentStorage=true targets the Azurite emulator above. + Storage__ConnectionString: "UseDevelopmentStorage=true" + Storage__QueueName: "integration-test-queue" + run: dotnet test --no-build --verbosity normal -p:ParallelizeTestCollections=false \ No newline at end of file diff --git a/.github/workflows/nuget.yml b/.github/workflows/nuget.yml index 2d7a1a7..7a174cc 100644 --- a/.github/workflows/nuget.yml +++ b/.github/workflows/nuget.yml @@ -1,39 +1,39 @@ -name: Nuget - -on: - push: - tags: - - 'v*' - -env: - PACKAGE_OUTPUT: ${{ github.workspace }}/package - NUGET_URL: https://api.nuget.org/v3/index.json - -jobs: - deploy: - name: Deploy - - runs-on: ubuntu-latest - - steps: - - uses: actions/checkout@v3 - - name: Setup .NET - uses: actions/setup-dotnet@v3 - with: - dotnet-version: | - 8.0.x - - name: Restore dependencies - run: dotnet restore - - - name: Build - run: dotnet build --no-restore --configuration Release - - - name: Get Version - id: version - uses: battila7/get-version-action@v2 - - - name: Package Project - run: dotnet pack --no-restore --no-build --configuration Release --include-symbols -p:PackageVersion=${{ steps.version.outputs.version-without-v }} --output ${{ env.PACKAGE_OUTPUT }} - - - name: Publish Package - run: dotnet nuget push ${{ env.PACKAGE_OUTPUT }}/*.nupkg -k ${{ secrets.NUGET_KEY }} -s ${{ env.NUGET_URL }} --skip-duplicate \ No newline at end of file +name: Nuget + +on: + push: + tags: + - 'v*' + +env: + PACKAGE_OUTPUT: ${{ github.workspace }}/package + NUGET_URL: https://api.nuget.org/v3/index.json + +jobs: + deploy: + name: Deploy + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + - name: Setup .NET + uses: actions/setup-dotnet@v4 + with: + dotnet-version: | + 8.0.x + - name: Restore dependencies + run: dotnet restore + + - name: Build + run: dotnet build --no-restore --configuration Release + + - name: Get Version + id: version + uses: battila7/get-version-action@v2 + + - name: Package Project + run: dotnet pack --no-restore --no-build --configuration Release --include-symbols -p:PackageVersion=${{ steps.version.outputs.version-without-v }} --output ${{ env.PACKAGE_OUTPUT }} + + - name: Publish Package + run: dotnet nuget push ${{ env.PACKAGE_OUTPUT }}/*.nupkg -k ${{ secrets.NUGET_KEY }} -s ${{ env.NUGET_URL }} --skip-duplicate \ No newline at end of file diff --git a/README.md b/README.md index a727d8c..17e5c9f 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,9 @@ # Azure Storage Queue Service library -Logo + + + Unifinity + [![.NET](https://github.com/jasonshave/JasonShave.AzureStorage.QueueService/actions/workflows/dotnet.yml/badge.svg)](https://github.com/jasonshave/JasonShave.AzureStorage.QueueService/actions/workflows/dotnet.yml) [![Nuget](https://github.com/jasonshave/JasonShave.AzureStorage.QueueService/actions/workflows/nuget.yml/badge.svg)](https://github.com/jasonshave/JasonShave.AzureStorage.QueueService/actions/workflows/nuget.yml) @@ -19,6 +22,9 @@ You will need to create an Azure Storage account in the Azure portal using a uni - Sending a message will automatically serialize the payload to `BinaryData`. - By default one message is pulled when `ReceiveMessagesAsync` is called, however this is configurable. - If your handler does not throw, messages are automatically removed from the queue otherwise the message is returned to the queue for delivery again. +- `ReceiveMessagesAsync` never throws for failures in the queue read, the deserialization, your handler, the delete, or your exception handler. Every failure is logged and passed to your exception handler, and the remaining messages in the batch are still processed. The only exception that escapes is `OperationCanceledException` for your own cancellation token, so a transient storage error cannot take down your host. +- A received message is hidden from other consumers for the *visibility timeout* (30 seconds unless you set `VisibilityTimeout`). If your handler runs longer than that, the message becomes visible again, another consumer can pick it up, and the delete on this side fails and is reported to your exception handler. Set `VisibilityTimeout` longer than your slowest handler multiplied by the number of messages you receive per call. +- A message that can never be processed (for example, invalid JSON) is redelivered every time its visibility timeout expires. Watch for repeated `DeserializationException`s in your exception handler and in the log, which includes the message's dequeue count. - Automatic deserialization uses the `System.Text.Json` deserialization behavior. This can be overridden by specifying your own `JsonSerializerOptions` as seen below. - You can 'peek' messages using `PeekMessages` which returns a collection but doesn't remove them from the queue. @@ -32,7 +38,8 @@ You will need to create an Azure Storage account in the Azure portal using a uni "QueueClientSettings" : { "ConnectionString": "[your_connection_string]", "QueueName": "[your_queue_name]", - "CreateIfNotExists": true + "CreateIfNotExists": true, + "VisibilityTimeout": "00:02:00" } } ``` @@ -56,7 +63,20 @@ You will need to create an Azure Storage account in the Azure portal using a uni })); ``` -4. You can create your queue in advance or allow the library to create it during runtime by setting the `CreateIfNotExists` property to `true`. +4. You can create your queue in advance or allow the library to create it during runtime by setting the `CreateIfNotExists` property to `true`. The queue is created the first time the client is used, not when it is resolved from dependency injection. + +### Per-client settings + +Every client has its own `QueueClientSettings`: + +| Setting | What it does | +|---|---| +| `ConnectionString` + `QueueName` | Connect with a storage account connection string. | +| `EndpointUri` + `TokenCredential` | Connect with an identity instead. The URI includes the queue name. | +| `CreateIfNotExists` | Create the queue on first use if it does not exist. | +| `VisibilityTimeout` | How long a received message is hidden from other consumers while your handler runs. Defaults to 30 seconds. | +| `SerializerOptions` | `JsonSerializerOptions` for this client only, used for both sending and receiving. | +| `ClientOptions` | `QueueClientOptions` for the underlying Azure SDK client: retry policy, diagnostics, and `MessageEncoding`. Set `MessageEncoding = QueueMessageEncoding.Base64` for queues shared with Azure Functions or older SDKs. | ## Configure dependency injection @@ -98,6 +118,10 @@ services.AddAzureStorageQueueClient(x => // get configuration from IConfiguration binder services.AddAzureStorageQueueClient(x => x.AddClient("MyClient1", y => Configuration.Bind(nameof(QueueClientSettings), y))); + +// or bind a configuration section directly +services.AddAzureStorageQueueClient(x => + x.AddClient("MyClient1", Configuration.GetSection("Queues:MyClient1"))); ``` ```csharp @@ -134,7 +158,7 @@ public class MyClass ### Example 2: Get a named client ```csharp -// inject the IQueueClientFactory and get a default client +// inject the IQueueClientFactory and get a named client public class MyClass { private readonly AzureStorageQueueClient _queueClient; @@ -146,26 +170,43 @@ public class MyClass } ``` +### Example 3: Inject a client directly + +Registered clients are also available from dependency injection without going through the factory. The default client is a plain singleton; named clients are keyed services. + +```csharp +public class MyClass +{ + public MyClass( + IAzureStorageQueueClient defaultClient, + [FromKeyedServices("MyClient1")] IAzureStorageQueueClient namedClient) + { + } +} +``` + +Depend on `IAzureStorageQueueClient` rather than the concrete `AzureStorageQueueClient` if you want to substitute a fake in your own unit tests. + ## Sending messages to an Azure storage account queue -The following example shows the .NET Worker Service template where the class uses the `IHostedService` interface to send a message every five seconds. +The following example shows the .NET Worker Service template where a `BackgroundService` sends a message every five seconds. Use `BackgroundService.ExecuteAsync` for loops like this rather than `IHostedService.StartAsync`: the host waits for `StartAsync` to return before it finishes starting, so a loop there blocks startup. 1. Inject the `IQueueClientFactory` interface and use as follows: ```csharp - public class Sender : IHostedService + public class Sender : BackgroundService { private readonly AzureStorageQueueClient _queueClient; public Sender(IQueueClientFactory queueClientFactory) => _queueClient = queueClientFactory.GetQueueClient(); - public async Task StartAsync(CancellationToken cancellationToken) + protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - while (!cancellationToken.IsCancellationRequested) + while (!stoppingToken.IsCancellationRequested) { var myMessage = new MyMessage("Test"); - await _queueClient.SendMessageAsync(myMessage, cancellationToken); - await Task.Delay(5000); + await _queueClient.SendMessageAsync(myMessage, cancellationToken: stoppingToken); + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); } } } @@ -173,32 +214,48 @@ The following example shows the .NET Worker Service template where the class use ## Receiving and handling messages from an Azure storage account queue -The following example shows the .NET Worker Service template where the class uses the `IHostedService` interface to run a particular code block repeatedly. The application will receive the payload from the queue repeatedly. +The following example shows the .NET Worker Service template where a `BackgroundService` polls the queue repeatedly. Your message handler receives the deserialized payload and the queue's metadata; your exception handler receives anything that went wrong while receiving, deserializing, handling or deleting a message. + +`ReceiveMessagesAsync` reports failures through your exception handler instead of throwing, so the loop below keeps running through transient storage errors. The `try`/`catch` is still worth having: it protects you from anything thrown by your own code outside the handlers, and an unhandled exception in `ExecuteAsync` stops the whole host by default in .NET 6 and later. 1. Inject the `IQueueClientFactory` interface and use as follows: ```csharp - public class MySubscriber : IHostedService + public class MySubscriber : BackgroundService { private readonly AzureStorageQueueClient _queueClient; - private readonly IMyMessageHandler _myMessageHandler; // see optional handler below + private readonly IMyMessageHandler _myMessageHandler; // your own handler + private readonly ILogger _logger; - public MySubscriber(IQueueClientFactory queueClientFactory, IMyMessageHandler myMessageHandler) + public MySubscriber(IQueueClientFactory queueClientFactory, IMyMessageHandler myMessageHandler, ILogger logger) { // get the default queue client _queueClient = queueClientFactory.GetQueueClient(); _myMessageHandler = myMessageHandler; + _logger = logger; } - public async Task StartAsync(CancellationToken cancellationToken) + protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - while (!cancellationToken.IsCancellationRequested) + while (!stoppingToken.IsCancellationRequested) { - await _queueClient.ReceiveMessagesAsync( - message => _myMessageHandler.HandleAsync(message), - exception => _myMessageHandler.HandleExceptionAsync(exception), - cancellationToken); - await Task.Delay(1000); + try + { + await _queueClient.ReceiveMessagesAsync( + (message, metadata) => _myMessageHandler.HandleAsync(message), + (exception, metadata) => _myMessageHandler.HandleExceptionAsync(exception), + cancellationToken: stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + _logger.LogError(ex, "Unexpected error while polling the queue."); + } + + await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken); } } } @@ -206,10 +263,10 @@ The following example shows the .NET Worker Service template where the class use ### Handling multiple messages -The library allows you to pull multiple messages by specifying the `maxMessage` count as an integer in the `ReceiveMessagesAsync()` method. These are sent to the handler as individual messages but pulled from the queue as a batch the consuming application would hold a lock on for the default duration used in the Azure Storage Queue library. +The library allows you to pull multiple messages by specifying the `numMessages` count in the `ReceiveMessagesAsync()` method (the service allows at most 32). These are sent to the handler one at a time but pulled from the queue as a batch, and every message in the batch is hidden from other consumers for the visibility timeout from the moment the batch is received. Set `VisibilityTimeout` so that it covers the whole batch, otherwise the last messages in the batch may become visible again before you reach them. ```csharp -await _queueClient.ReceiveMessagesAsync(HandleMessage, HandleException, cancellationToken, 10); +await _queueClient.ReceiveMessagesAsync(HandleMessage, HandleException, numMessages: 10, cancellationToken: stoppingToken); ``` ## OpenTelemetry Support @@ -240,8 +297,25 @@ The following metrics have also been configured and read through tools like Azur - `queue_messages_received_total` - `queue_messages_sent_total` - `queue_messages_processed_total` -- `queue_message_processing_duration_seconds` +- `queue_messages_failed_total` +- `queue_message_processing_duration_seconds` (tagged with `queue.process_success`) +- `queue_message_size_bytes` + +All metrics are tagged with `queue.name`, so clients for different queues can be told apart. + +## Upgrading to 2.1 + +No code changes are required, but note these behaviour changes: + +- `ReceiveMessagesAsync` no longer throws when the queue read fails, when your exception handler throws, or when a message cannot be deleted. All of these now go to your exception handler (with `null` metadata when the read itself failed) and are logged. If you relied on catching a failed read in your own loop, that catch block will no longer fire. +- A queue configured with `CreateIfNotExists` is created on the first send, receive or peek instead of when the client is resolved from dependency injection, so startup no longer makes a blocking network call. +- Client configuration delegates run when the client is first requested rather than when `AddAzureStorageQueueClient` is called, and registering the same client name twice now adds to its configuration instead of ignoring the second registration. +- Custom `JsonSerializerOptions` are now applied when sending messages as well as when receiving them. If you configured a naming policy, messages you send will change shape accordingly. +- `Azure.Identity` is no longer a dependency of this package. Add it to your own project if you use `DefaultAzureCredential`. +- The package now depends on `Azure.Storage.Queues` 12.28 and `Microsoft.Extensions.*` 10.0. The 10.0 packages still target .NET 8, so no framework change is needed, but any `Microsoft.Extensions.*` packages you reference directly will be raised to 10.0 as well. This floor comes from the Azure SDK itself (`Azure.Core` 1.60 requires it), so it applies to any project on the current Azure SDK. ## License This project is licensed under the MIT License - see the [LICENSE.md](license.md) file for details. + +The logo is the Unifinity Software Corporation brand mark, taken unchanged from the Unifinity brand kit. It is not covered by the MIT licence. diff --git a/images/logo-dark.svg b/images/logo-dark.svg new file mode 100644 index 0000000..a4d66e2 --- /dev/null +++ b/images/logo-dark.svg @@ -0,0 +1,5 @@ + +Unifinity + + + diff --git a/images/logo-light.svg b/images/logo-light.svg new file mode 100644 index 0000000..b7ca51c --- /dev/null +++ b/images/logo-light.svg @@ -0,0 +1,5 @@ + +Unifinity + + + diff --git a/images/logo.jpg b/images/logo.jpg deleted file mode 100644 index c61df8d..0000000 Binary files a/images/logo.jpg and /dev/null differ diff --git a/images/logo.png b/images/logo.png new file mode 100644 index 0000000..52a1c4b Binary files /dev/null and b/images/logo.png differ diff --git a/src/AzureStorage.QueueService/AzureStorage.QueueService.csproj b/src/AzureStorage.QueueService/AzureStorage.QueueService.csproj index 5a1f981..10a8386 100644 --- a/src/AzureStorage.QueueService/AzureStorage.QueueService.csproj +++ b/src/AzureStorage.QueueService/AzureStorage.QueueService.csproj @@ -14,7 +14,7 @@ Unifinity Software Corporation Azure, Azure Storage, Azure Storage Queue, Storage Queue, Queue Service, Jason Shave, Unifinity Software Corporation https://github.com/jasonshave/JasonShave.AzureStorage.QueueService - logo.jpg + logo.png README.md git https://github.com/jasonshave/JasonShave.AzureStorage.QueueService @@ -24,13 +24,12 @@ - - - - - - - + + + + + + @@ -47,7 +46,7 @@ - + True \ diff --git a/src/AzureStorage.QueueService/AzureStorageQueueClient.cs b/src/AzureStorage.QueueService/AzureStorageQueueClient.cs index fa2c929..5f160fb 100644 --- a/src/AzureStorage.QueueService/AzureStorageQueueClient.cs +++ b/src/AzureStorage.QueueService/AzureStorageQueueClient.cs @@ -7,104 +7,222 @@ namespace AzureStorage.QueueService; -public class AzureStorageQueueClient +public class AzureStorageQueueClient : IAzureStorageQueueClient { + // The service default when no visibility timeout is supplied on receive. + private static readonly TimeSpan DefaultVisibilityTimeout = TimeSpan.FromSeconds(30); + private readonly IMessageConverter _messageConverter; private readonly QueueClient _queueClient; private readonly ILogger _logger; private readonly QueueServiceTelemetrySettings _telemetrySettings; + private readonly TimeSpan? _visibilityTimeout; + private readonly bool _createIfNotExists; + private readonly SemaphoreSlim _createLock = new(1, 1); + private volatile bool _queueEnsured; internal AzureStorageQueueClient( - IMessageConverter messageConverter, - QueueClient queueClient, + IMessageConverter messageConverter, + QueueClient queueClient, ILoggerFactory loggerFactory, - IOptions telemetrySettingsOptions) + IOptions telemetrySettingsOptions, + TimeSpan? visibilityTimeout = null, + bool createIfNotExists = false) { _messageConverter = messageConverter; _queueClient = queueClient; _logger = loggerFactory.CreateLogger(); _telemetrySettings = telemetrySettingsOptions.Value; + _visibilityTimeout = visibilityTimeout; + _createIfNotExists = createIfNotExists; } - public async ValueTask CreateQueueIfNotExistsAsync(IDictionary? metadata = null, CancellationToken cancellationToken = default) => + public string QueueName => _queueClient.Name; + + public async ValueTask CreateQueueIfNotExistsAsync(IDictionary? metadata = null, CancellationToken cancellationToken = default) + { await _queueClient.CreateIfNotExistsAsync(metadata, cancellationToken); + _queueEnsured = true; + } - public async ValueTask ClearMessagesAsync(CancellationToken cancellationToken = default) => + public async ValueTask ClearMessagesAsync(CancellationToken cancellationToken = default) + { + await EnsureQueueExistsAsync(cancellationToken); await _queueClient.ClearMessagesAsync(cancellationToken); + } - public async ValueTask> PeekMessagesAsync(int numMessages, CancellationToken cancellationToken = default) => - (await _queueClient.PeekMessagesAsync(numMessages, cancellationToken)).Value.Convert(_messageConverter); + public async ValueTask> PeekMessagesAsync(int numMessages, CancellationToken cancellationToken = default) + { + await EnsureQueueExistsAsync(cancellationToken); + return (await _queueClient.PeekMessagesAsync(numMessages, cancellationToken)).Value.Convert(_messageConverter); + } - public IEnumerable PeekMessages(int numMessages, CancellationToken cancellationToken = default) => - _queueClient.PeekMessages(numMessages, cancellationToken).Value.Convert(_messageConverter); + public IEnumerable PeekMessages(int numMessages, CancellationToken cancellationToken = default) + { + EnsureQueueExists(cancellationToken); + return _queueClient.PeekMessages(numMessages, cancellationToken).Value.Convert(_messageConverter); + } /// - /// Receives a message of the type specified and deserializes the input using a JSON message converter. - /// The message is processed using a function delegate and errors are processed using an exception handling delegate. + /// Receives up to messages, deserializes each one and passes it to + /// . A message is deleted from the queue only after its handler completes; + /// if the handler throws, the message is left on the queue and becomes visible again once its visibility + /// timeout expires. /// - /// - /// - /// - /// - /// - /// - /// - public async ValueTask ReceiveMessagesAsync(Func?, ValueTask> handleMessage, Func?, ValueTask> handleException, int numMessages = 1, CancellationToken cancellationToken = default) + /// + /// This method does not throw for failures in the queue read, the message conversion, the handler, the delete, + /// or itself. All of them are logged and passed to + /// (with null metadata when the read itself failed), and processing + /// continues with the next message in the batch. The only exception that escapes is + /// for the caller's own . + /// + /// The type to deserialize each message into. + /// Called once per message with the deserialized payload and the queue's metadata. + /// Called when receiving, deserializing, handling or deleting a message fails. + /// How many messages to receive in one call (the service allows at most 32). + /// Stops the receive and any further message processing. Messages still in the batch become visible again after their visibility timeout. + public async ValueTask ReceiveMessagesAsync( + Func?, ValueTask> handleMessage, + Func?, ValueTask> handleException, + int numMessages = 1, + CancellationToken cancellationToken = default) where TMessage : class { + ArgumentNullException.ThrowIfNull(handleMessage); + ArgumentNullException.ThrowIfNull(handleException); + using Activity? activity = _telemetrySettings.CreateNewActivityOnMessageRetrieval ? QueueServiceDiagnostics.Activities.StartReadingQueueActivity(tags => { - tags.Add(QueueServiceDiagnostics.Names.QueueName, _queueClient.Name); + tags.Add(QueueServiceDiagnostics.Names.QueueName, QueueName); tags.Add(QueueServiceDiagnostics.Names.QueueMessageCount, numMessages); }) : null; - QueueMessage[] receivedMessages = await _queueClient.ReceiveMessagesAsync(numMessages, null, cancellationToken); + QueueMessage[] receivedMessages; + try + { + await EnsureQueueExistsAsync(cancellationToken); + receivedMessages = await _queueClient.ReceiveMessagesAsync(numMessages, _visibilityTimeout, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + // A failed read (network, auth, missing queue, bad numMessages) used to escape straight into the + // caller's polling loop and take the host down. It is now reported like any other failure. + _logger.LogReceiveFailed(QueueName, ex); + activity?.AddException(ex); + await InvokeExceptionHandlerAsync(handleException, ex, metadata: null, messageId: null); + return; + } + + if (receivedMessages.Length == 0) return; + + _logger.LogMessagesReceived(QueueName, receivedMessages.Length); + activity?.AddEvent(new ActivityEvent(QueueServiceDiagnostics.Events.QueueMessageReceived)); + + var metadata = await TryGetMetadataAsync(cancellationToken); - if (receivedMessages.Any()) + foreach (var queueMessage in receivedMessages) { - _logger.LogMessageCount(receivedMessages.Length); - activity?.AddEvent(QueueServiceDiagnostics.Events.QueueMessageReceived); + cancellationToken.ThrowIfCancellationRequested(); + await ProcessMessageAsync(queueMessage, metadata, handleMessage, handleException, activity, cancellationToken); + } + } - var queueProperties = await _queueClient.GetPropertiesAsync(cancellationToken); + private async ValueTask ProcessMessageAsync( + QueueMessage queueMessage, + IDictionary? metadata, + Func?, ValueTask> handleMessage, + Func?, ValueTask> handleException, + Activity? activity, + CancellationToken cancellationToken) + where TMessage : class + { + var tags = new TagList + { + new(QueueServiceDiagnostics.Names.QueueName, QueueName), + }; - foreach (var queueMessage in receivedMessages) - { - // Tags common to all metrics for this message - var tagList = new TagList() - { - new(QueueServiceDiagnostics.Names.QueueName, _queueClient.Name), - }; - - // 1) Increment "messages received" count - QueueServiceDiagnostics.Metrics.MessagesReceived.Add(1, tagList); - - // 2) Measure how long it takes to process + handle failures - await QueueServiceDiagnostics.Metrics.MessageProcessingDuration.ProcessWithMetricsAsync( - async () => - { - var convertedMessage = _messageConverter.Convert(queueMessage.MessageText); - await handleMessage(convertedMessage, queueProperties?.Value.Metadata); - - _logger.LogProcessedMessage(queueMessage.MessageId); - await _queueClient.DeleteMessageAsync(queueMessage.MessageId, queueMessage.PopReceipt, cancellationToken); - - // 3) Add an "message processed" event for your Activity - activity?.AddEvent(QueueServiceDiagnostics.Events.QueueMessageProcessed); - - // 4) Increment "messages processed" count - QueueServiceDiagnostics.Metrics.MessagesProcessed.Add(1, tagList); - }, - tagList, - onFailure: async ex => - { - // Custom error logging / handling - activity?.AddException(ex); - await handleException(ex, queueProperties?.Value.Metadata); - } - ); - } + QueueServiceDiagnostics.Metrics.MessagesReceived.Add(1, tags); + + var start = Stopwatch.GetTimestamp(); + var handled = false; + var success = false; + + try + { + var convertedMessage = _messageConverter.Convert(queueMessage.MessageText); + await handleMessage(convertedMessage, metadata); + handled = true; + + // The handler's work is done, so the delete is deliberately not tied to the caller's token: + // cancelling it during shutdown would only redeliver a message that was already processed. + await _queueClient.DeleteMessageAsync(queueMessage.MessageId, queueMessage.PopReceipt, CancellationToken.None); + + _logger.LogMessageProcessed(QueueName, queueMessage.MessageId); + activity?.AddEvent(new ActivityEvent(QueueServiceDiagnostics.Events.QueueMessageProcessed)); + QueueServiceDiagnostics.Metrics.MessagesProcessed.Add(1, tags); + success = true; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + if (handled) + _logger.LogDeleteFailed(QueueName, queueMessage.MessageId, _visibilityTimeout ?? DefaultVisibilityTimeout, ex); + else + _logger.LogHandlerFailed(QueueName, queueMessage.MessageId, queueMessage.DequeueCount, ex); + + QueueServiceDiagnostics.Metrics.MessageFailures.Add(1, tags); + activity?.AddException(ex); + await InvokeExceptionHandlerAsync(handleException, ex, metadata, queueMessage.MessageId); + } + finally + { + tags.Add(QueueServiceDiagnostics.Names.QueueProcessSuccess, success); + QueueServiceDiagnostics.Metrics.MessageProcessingDuration.Record(Stopwatch.GetElapsedTime(start).TotalSeconds, tags); + } + } + + private async ValueTask InvokeExceptionHandlerAsync( + Func?, ValueTask> handleException, + Exception exception, + IDictionary? metadata, + string? messageId) + { + try + { + await handleException(exception, metadata); + } + catch (Exception handlerException) + { + // The caller's exception handler is the last line of defence; if it throws there is nowhere left + // to report to but the log. Letting it escape would abandon the rest of the batch. + _logger.LogExceptionHandlerFailed(QueueName, messageId ?? "(none)", handlerException); + } + } + + private async ValueTask?> TryGetMetadataAsync(CancellationToken cancellationToken) + { + try + { + var response = await _queueClient.GetPropertiesAsync(cancellationToken); + return response?.Value?.Metadata; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + _logger.LogMetadataFailed(QueueName, ex); + return null; } } @@ -113,33 +231,77 @@ await QueueServiceDiagnostics.Metrics.MessageProcessingDuration.ProcessWithMetri /// /// /// + /// + /// /// /// - /// + /// Serialization and send failures are logged and rethrown. public async ValueTask SendMessageAsync(TMessage message, TimeSpan? visibilityTimeout = null, TimeSpan? timeToLive = null, CancellationToken cancellationToken = default) { using Activity? activity = _telemetrySettings.CreateNewActivityOnMessageSend ? QueueServiceDiagnostics.Activities.StartSendingMessageActivity(tags => { - tags.Add(QueueServiceDiagnostics.Names.QueueName, _queueClient.Name); + tags.Add(QueueServiceDiagnostics.Names.QueueName, QueueName); }) : null; try { + await EnsureQueueExistsAsync(cancellationToken); + BinaryData binaryMessage = _messageConverter.Convert(message); SendReceipt response = await _queueClient.SendMessageAsync(binaryMessage, visibilityTimeout, timeToLive, cancellationToken); - QueueServiceDiagnostics.Metrics.MessagesSent.Add(1, new TagList() + + var tags = new TagList { - new(QueueServiceDiagnostics.Names.QueueName, _queueClient.Name), - }); + new(QueueServiceDiagnostics.Names.QueueName, QueueName), + }; + QueueServiceDiagnostics.Metrics.MessagesSent.Add(1, tags); + QueueServiceDiagnostics.Metrics.MessageSize.Record(binaryMessage.ToMemory().Length, tags); return response; } catch (Exception e) { - _logger.LogSendError(e.Message); + _logger.LogSendFailed(QueueName, e); + activity?.AddException(e); throw; } } -} \ No newline at end of file + + // Queue creation used to be a blocking network call made while resolving the client from DI, typically inside + // a constructor during startup. It now happens on the first operation, once, and asynchronously. + private async ValueTask EnsureQueueExistsAsync(CancellationToken cancellationToken) + { + if (!_createIfNotExists || _queueEnsured) return; + + await _createLock.WaitAsync(cancellationToken); + try + { + if (_queueEnsured) return; + await _queueClient.CreateIfNotExistsAsync(cancellationToken: cancellationToken); + _queueEnsured = true; + } + finally + { + _createLock.Release(); + } + } + + private void EnsureQueueExists(CancellationToken cancellationToken) + { + if (!_createIfNotExists || _queueEnsured) return; + + _createLock.Wait(cancellationToken); + try + { + if (_queueEnsured) return; + _queueClient.CreateIfNotExists(cancellationToken: cancellationToken); + _queueEnsured = true; + } + finally + { + _createLock.Release(); + } + } +} diff --git a/src/AzureStorage.QueueService/DeserializationException.cs b/src/AzureStorage.QueueService/DeserializationException.cs index 3f61c89..7ffbdfb 100644 --- a/src/AzureStorage.QueueService/DeserializationException.cs +++ b/src/AzureStorage.QueueService/DeserializationException.cs @@ -1,10 +1,18 @@ namespace AzureStorage.QueueService; +/// +/// Thrown when a queue message cannot be deserialized into the requested type. The original serializer error +/// is available as . +/// public class DeserializationException : Exception { public DeserializationException(string message) - : base(message) + : base(message) { + } + public DeserializationException(string message, Exception innerException) + : base(message, innerException) + { } -} \ No newline at end of file +} diff --git a/src/AzureStorage.QueueService/IAzureStorageQueueClient.cs b/src/AzureStorage.QueueService/IAzureStorageQueueClient.cs new file mode 100644 index 0000000..cab143b --- /dev/null +++ b/src/AzureStorage.QueueService/IAzureStorageQueueClient.cs @@ -0,0 +1,34 @@ +using Azure.Storage.Queues.Models; + +namespace AzureStorage.QueueService; + +/// +/// A client bound to one Azure Storage queue. is the only implementation; +/// depend on this interface when you want to substitute a fake in your own unit tests. +/// +public interface IAzureStorageQueueClient +{ + /// + /// The name of the queue this client is bound to. + /// + string QueueName { get; } + + ValueTask CreateQueueIfNotExistsAsync(IDictionary? metadata = null, CancellationToken cancellationToken = default); + + ValueTask ClearMessagesAsync(CancellationToken cancellationToken = default); + + ValueTask> PeekMessagesAsync(int numMessages, CancellationToken cancellationToken = default); + + IEnumerable PeekMessages(int numMessages, CancellationToken cancellationToken = default); + + /// + ValueTask ReceiveMessagesAsync( + Func?, ValueTask> handleMessage, + Func?, ValueTask> handleException, + int numMessages = 1, + CancellationToken cancellationToken = default) + where TMessage : class; + + /// + ValueTask SendMessageAsync(TMessage message, TimeSpan? visibilityTimeout = null, TimeSpan? timeToLive = null, CancellationToken cancellationToken = default); +} diff --git a/src/AzureStorage.QueueService/IQueueClientBuilder.cs b/src/AzureStorage.QueueService/IQueueClientBuilder.cs index 2cc446a..01cb4e5 100644 --- a/src/AzureStorage.QueueService/IQueueClientBuilder.cs +++ b/src/AzureStorage.QueueService/IQueueClientBuilder.cs @@ -4,7 +4,9 @@ namespace AzureStorage.QueueService; internal interface IQueueClientBuilder { - QueueClient CreateClient(QueueClientSettings settings); - - Task CreateClientAsync(QueueClientSettings settings); -} \ No newline at end of file + /// + /// Constructs the SDK client for the given settings. names the client + /// (for example "Queue client 'orders'") so configuration errors can say which one is at fault. + /// + QueueClient CreateClient(string clientDescription, QueueClientSettings settings); +} diff --git a/src/AzureStorage.QueueService/JsonQueueMessageConverter.cs b/src/AzureStorage.QueueService/JsonQueueMessageConverter.cs index e3a17fd..b2331ce 100644 --- a/src/AzureStorage.QueueService/JsonQueueMessageConverter.cs +++ b/src/AzureStorage.QueueService/JsonQueueMessageConverter.cs @@ -16,43 +16,61 @@ public JsonQueueMessageConverter(JsonSerializerOptions? serializerOptions = null { try { - var convertedMessage = JsonSerializer.Deserialize(input, _serializerOptions); - return convertedMessage; + return JsonSerializer.Deserialize(input, _serializerOptions); } catch (Exception e) { - throw new DeserializationException(e.Message); + throw new DeserializationException(Describe(e), e); } } public TOutput? Convert(string input) { - if (IsBase64(input)) + // Messages written by this library (and by the v12 SDK by default) are plain JSON; messages written by + // older SDKs, Azure Functions or the portal are Base64-encoded. Try the text as-is first so that short + // JSON values which also happen to be valid Base64 ("true", "null", "1234") are not mangled, and only + // then fall back to decoding. + try { - var byteData = System.Convert.FromBase64String(input); - input = Encoding.UTF8.GetString(byteData); + return JsonSerializer.Deserialize(input, _serializerOptions); } - - try + catch (JsonException) when (TryDecodeBase64(input, out var decoded)) { - var convertedMessage = JsonSerializer.Deserialize(input, _serializerOptions); - return convertedMessage; + try + { + return JsonSerializer.Deserialize(decoded, _serializerOptions); + } + catch (Exception e) + { + throw new DeserializationException(Describe(e), e); + } } catch (Exception e) { - throw new DeserializationException(e.Message); + throw new DeserializationException(Describe(e), e); } } public BinaryData Convert(TInput input) { - var binaryData = new BinaryData(input); - return binaryData; + // Honour the caller's serializer options on the way out as well as on the way in. + return BinaryData.FromObjectAsJson(input, _serializerOptions); } - private static bool IsBase64(string input) + private static string Describe(Exception e) => + $"Could not deserialize the queue message as {typeof(TOutput).Name}: {e.Message}"; + + private static bool TryDecodeBase64(string input, out string decoded) { - Span buffer = new Span(new byte[input.Length]); - return System.Convert.TryFromBase64String(input, buffer, out _); + // Decoded output is never longer than the encoded input. + var buffer = new byte[input.Length]; + if (System.Convert.TryFromBase64String(input, buffer, out var bytesWritten)) + { + decoded = Encoding.UTF8.GetString(buffer, 0, bytesWritten); + return true; + } + + decoded = string.Empty; + return false; } -} \ No newline at end of file +} diff --git a/src/AzureStorage.QueueService/LoggerMessageExtensions.cs b/src/AzureStorage.QueueService/LoggerMessageExtensions.cs index f3b7e01..cc871f8 100644 --- a/src/AzureStorage.QueueService/LoggerMessageExtensions.cs +++ b/src/AzureStorage.QueueService/LoggerMessageExtensions.cs @@ -4,12 +4,39 @@ namespace AzureStorage.QueueService; public static partial class LoggerMessageExtensions { + [Obsolete("Use LogMessageProcessed, which includes the queue name. This method will be removed in 3.0.")] [LoggerMessage(0, LogLevel.Information, "Processed message successfully, removing queue message id: {MessageId}")] public static partial void LogProcessedMessage(this ILogger logger, string messageId); + [Obsolete("Use LogMessagesReceived, which includes the queue name. This method will be removed in 3.0.")] [LoggerMessage(1, LogLevel.Information, "Processing {MessageCount} storage queue message(s).")] public static partial void LogMessageCount(this ILogger logger, int messageCount); + [Obsolete("Use LogSendFailed, which includes the queue name and the exception. This method will be removed in 3.0.")] [LoggerMessage(2, LogLevel.Error, "There was a problem sending the message. The error was: {ErrorMessage}")] public static partial void LogSendError(this ILogger logger, string errorMessage); -} \ No newline at end of file + + [LoggerMessage(3, LogLevel.Information, "Queue '{QueueName}': processed message {MessageId} successfully; removing it from the queue.")] + public static partial void LogMessageProcessed(this ILogger logger, string? queueName, string messageId); + + [LoggerMessage(4, LogLevel.Information, "Queue '{QueueName}': processing {MessageCount} message(s).")] + public static partial void LogMessagesReceived(this ILogger logger, string? queueName, int messageCount); + + [LoggerMessage(5, LogLevel.Error, "Queue '{QueueName}': there was a problem sending the message.")] + public static partial void LogSendFailed(this ILogger logger, string? queueName, Exception exception); + + [LoggerMessage(6, LogLevel.Error, "Queue '{QueueName}': failed to receive messages. The error has been passed to the exception handler.")] + public static partial void LogReceiveFailed(this ILogger logger, string? queueName, Exception exception); + + [LoggerMessage(7, LogLevel.Error, "Queue '{QueueName}': the message handler threw for message {MessageId} (dequeue count {DequeueCount}). The message was not deleted and will become visible again.")] + public static partial void LogHandlerFailed(this ILogger logger, string? queueName, string messageId, long dequeueCount, Exception exception); + + [LoggerMessage(8, LogLevel.Warning, "Queue '{QueueName}': message {MessageId} was handled successfully but could not be deleted. The handler probably ran longer than the visibility timeout ({VisibilityTimeout}), so the message may be delivered again.")] + public static partial void LogDeleteFailed(this ILogger logger, string? queueName, string messageId, TimeSpan visibilityTimeout, Exception exception); + + [LoggerMessage(9, LogLevel.Error, "Queue '{QueueName}': the exception handler itself threw while handling message {MessageId}. Continuing with the next message.")] + public static partial void LogExceptionHandlerFailed(this ILogger logger, string? queueName, string messageId, Exception exception); + + [LoggerMessage(10, LogLevel.Warning, "Queue '{QueueName}': could not read the queue metadata; the handler will receive null metadata.")] + public static partial void LogMetadataFailed(this ILogger logger, string? queueName, Exception exception); +} diff --git a/src/AzureStorage.QueueService/QueueClientBuilder.cs b/src/AzureStorage.QueueService/QueueClientBuilder.cs index cc54a22..f7b14de 100644 --- a/src/AzureStorage.QueueService/QueueClientBuilder.cs +++ b/src/AzureStorage.QueueService/QueueClientBuilder.cs @@ -4,38 +4,21 @@ namespace AzureStorage.QueueService; internal class QueueClientBuilder : IQueueClientBuilder { - public QueueClient CreateClient(QueueClientSettings settings) + public QueueClient CreateClient(string clientDescription, QueueClientSettings settings) { - var client = Create(settings); + if (!string.IsNullOrEmpty(settings.ConnectionString)) + { + if (string.IsNullOrEmpty(settings.QueueName)) + throw new ApplicationException($"{clientDescription} has a ConnectionString but no QueueName."); - if (settings.CreateIfNotExists) client.CreateIfNotExists(); + return new QueueClient(settings.ConnectionString, settings.QueueName, settings.ClientOptions); + } - return client; - } - - public async Task CreateClientAsync(QueueClientSettings settings) - { - var client = Create(settings); - - if (settings.CreateIfNotExists) await client.CreateIfNotExistsAsync(); - - return client; - } - - private static QueueClient Create(QueueClientSettings settings) - { - QueueClient? client = default; - if (settings.TokenCredential is not null && settings.EndpointUri is not null) - client = new QueueClient(settings.EndpointUri, settings.TokenCredential); - - if (settings.ConnectionString is not null) - client = new QueueClient(settings.ConnectionString, settings.QueueName); - - if (client is null) - throw new ApplicationException( - "There was a problem creating the client. Unable to determine if the endpoint URI or connection string should be used. " + - "Please be sure to set the connection string or the token credential and endpoint URI together."); + if (settings.EndpointUri is not null && settings.TokenCredential is not null) + return new QueueClient(settings.EndpointUri, settings.TokenCredential, settings.ClientOptions); - return client; + throw new ApplicationException( + $"{clientDescription} is not configured. Set ConnectionString and QueueName, " + + "or set EndpointUri (including the queue name) and TokenCredential."); } -} \ No newline at end of file +} diff --git a/src/AzureStorage.QueueService/QueueClientFactory.cs b/src/AzureStorage.QueueService/QueueClientFactory.cs index 1018622..a7010df 100644 --- a/src/AzureStorage.QueueService/QueueClientFactory.cs +++ b/src/AzureStorage.QueueService/QueueClientFactory.cs @@ -1,4 +1,5 @@ -using AzureStorage.QueueService.Telemetry; +using System.Collections.Concurrent; +using AzureStorage.QueueService.Telemetry; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -6,22 +7,25 @@ namespace AzureStorage.QueueService; internal sealed class QueueClientFactory : IQueueClientFactory { - private readonly Dictionary _namedClients = new(); - private AzureStorageQueueClient? _defaultClient; + // Lazy so two threads asking for the same client for the first time get one instance, not two. + private readonly ConcurrentDictionary> _clients = new(StringComparer.Ordinal); private readonly QueueClientSettingsRegistry _registry; + private readonly IOptionsMonitor _settings; private readonly IQueueClientBuilder _queueClientBuilder; private readonly ILoggerFactory _loggerFactory; private readonly IMessageConverter _messageConverter; private readonly IOptions _telemetrySettings; public QueueClientFactory( - QueueClientSettingsRegistry registry, - IQueueClientBuilder queueClientBuilder, - ILoggerFactory loggerFactory, + QueueClientSettingsRegistry registry, + IOptionsMonitor settings, + IQueueClientBuilder queueClientBuilder, + ILoggerFactory loggerFactory, IMessageConverter messageConverter, IOptions telemetrySettingsOptions) { _registry = registry; + _settings = settings; _queueClientBuilder = queueClientBuilder; _loggerFactory = loggerFactory; _messageConverter = messageConverter; @@ -30,38 +34,53 @@ public QueueClientFactory( public AzureStorageQueueClient GetQueueClient(string clientName) { - // try named client - _namedClients.TryGetValue(clientName, out var azureStorageQueueClient); - if (azureStorageQueueClient is not null) - { - return azureStorageQueueClient; - } + ArgumentNullException.ThrowIfNull(clientName); - // not found so create one and add it - _registry.NamedClientsSettings.TryGetValue(clientName, out var customClientSettings); - if (customClientSettings is null) + if (!_registry.Contains(clientName)) { - throw new ApplicationException($"Settings for named client, {clientName} not found."); - } + var registered = _registry.ClientNames.Count == 0 + ? "none" + : string.Join(", ", _registry.ClientNames.Select(name => $"'{name}'")); - var client = Create(customClientSettings); - _namedClients.TryAdd(clientName, client); + throw new ApplicationException( + $"No queue client named '{clientName}' has been registered (registered named clients: {registered}). " + + $"Register it with services.AddAzureStorageQueueClient(x => x.AddClient(\"{clientName}\", ...))."); + } - return client; + return GetOrCreate(clientName); } public AzureStorageQueueClient GetQueueClient() { - // use default client - if (_defaultClient is not null) return _defaultClient; - _defaultClient = Create(_registry.DefaultClientSettings); - return _defaultClient; + if (!_registry.HasDefaultClient) + { + throw new ApplicationException( + "No default queue client has been registered. Register one with " + + "services.AddAzureStorageQueueClient(x => x.AddDefaultClient(...)), or ask for a named client with GetQueueClient(name)."); + } + + return GetOrCreate(Options.DefaultName); } - private AzureStorageQueueClient Create(QueueClientSettings settings) + private AzureStorageQueueClient GetOrCreate(string clientName) => + _clients.GetOrAdd(clientName, name => new Lazy(() => Create(name), LazyThreadSafetyMode.ExecutionAndPublication)).Value; + + private AzureStorageQueueClient Create(string clientName) { - var queueClient = _queueClientBuilder.CreateClient(settings); - var azureStorageQueueClient = new AzureStorageQueueClient(_messageConverter, queueClient, _loggerFactory, _telemetrySettings); - return azureStorageQueueClient; + var settings = _settings.Get(clientName); + var description = clientName == Options.DefaultName ? "The default queue client" : $"Queue client '{clientName}'"; + var queueClient = _queueClientBuilder.CreateClient(description, settings); + + var messageConverter = settings.SerializerOptions is null + ? _messageConverter + : new JsonQueueMessageConverter(settings.SerializerOptions); + + return new AzureStorageQueueClient( + messageConverter, + queueClient, + _loggerFactory, + _telemetrySettings, + settings.VisibilityTimeout, + settings.CreateIfNotExists); } -} \ No newline at end of file +} diff --git a/src/AzureStorage.QueueService/QueueClientSettings.cs b/src/AzureStorage.QueueService/QueueClientSettings.cs index 97f33be..43d0c07 100644 --- a/src/AzureStorage.QueueService/QueueClientSettings.cs +++ b/src/AzureStorage.QueueService/QueueClientSettings.cs @@ -1,5 +1,7 @@ using System.Diagnostics.CodeAnalysis; +using System.Text.Json; using Azure.Core; +using Azure.Storage.Queues; namespace AzureStorage.QueueService; @@ -8,13 +10,43 @@ public class QueueClientSettings { public const string SectionName = "AzureStorageQueueClientSettings"; + /// + /// The storage account connection string. Use this together with , + /// or use and instead. + /// public string? ConnectionString { get; set; } public string? QueueName { get; set; } + /// + /// When true, the queue is created the first time the client is used if it does not already exist. + /// public bool CreateIfNotExists { get; set; } - + + /// + /// The full queue URI, including the queue name. Use this together with . + /// public Uri? EndpointUri { get; set; } public TokenCredential? TokenCredential { get; set; } -} \ No newline at end of file + + /// + /// How long a received message stays hidden from other consumers while your handler runs. When not set, the + /// service default of 30 seconds applies. If a handler runs longer than this the message becomes visible again, + /// another consumer can pick it up, and this consumer's delete fails. Allow for the slowest handler multiplied + /// by the number of messages received per call, since a batch is processed one message at a time. + /// + public TimeSpan? VisibilityTimeout { get; set; } + + /// + /// JSON serializer options for this client only. When set, they take precedence over the options passed to + /// AddAzureStorageQueueClient for both sending and receiving. + /// + public JsonSerializerOptions? SerializerOptions { get; set; } + + /// + /// Options for the underlying Azure SDK client, such as the retry policy or + /// for queues that carry Base64-encoded messages. + /// + public QueueClientOptions? ClientOptions { get; set; } +} diff --git a/src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs b/src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs index 2c776e0..d79249d 100644 --- a/src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs +++ b/src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs @@ -1,27 +1,74 @@ -namespace AzureStorage.QueueService; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Options; + +namespace AzureStorage.QueueService; public class QueueClientSettingsBuilder { + internal IServiceCollection Services { get; } + internal QueueClientSettingsRegistry Registry { get; } - internal QueueClientSettingsBuilder(QueueClientSettingsRegistry registry) + internal QueueClientSettingsBuilder(IServiceCollection services, QueueClientSettingsRegistry registry) { + Services = services; Registry = registry; } - public QueueClientSettingsBuilder AddClient(string clientName, Action settings) + /// + /// Registers a named client. Get it from or inject it + /// directly with [FromKeyedServices("name")] AzureStorageQueueClient client. Registering the same name + /// again adds to its configuration rather than replacing it. + /// + public QueueClientSettingsBuilder AddClient(string clientName, Action configure) { - var queueClientSettings = new QueueClientSettings(); - settings(queueClientSettings); + ArgumentException.ThrowIfNullOrEmpty(clientName); + ArgumentNullException.ThrowIfNull(configure); - Registry.NamedClientsSettings.TryAdd(clientName, queueClientSettings); + Registry.Add(clientName); + Services.Configure(clientName, configure); + Services.TryAddKeyedSingleton(clientName, static (provider, key) => + provider.GetRequiredService().GetQueueClient((string)key!)); + Services.TryAddKeyedSingleton(clientName, static (provider, key) => + provider.GetRequiredKeyedService(key)); return this; } - public void AddDefaultClient(Action settings) + /// + /// Registers a named client bound from a configuration section, for example + /// configuration.GetSection("Queues:Orders"). + /// + public QueueClientSettingsBuilder AddClient(string clientName, IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(configuration); + return AddClient(clientName, settings => configuration.Bind(settings)); + } + + /// + /// Registers the default client. Get it from or inject + /// (or ) directly. + /// + public void AddDefaultClient(Action configure) + { + ArgumentNullException.ThrowIfNull(configure); + + Registry.HasDefaultClient = true; + Services.Configure(Options.DefaultName, configure); + Services.TryAddSingleton(static provider => + provider.GetRequiredService().GetQueueClient()); + Services.TryAddSingleton(static provider => + provider.GetRequiredService()); + } + + /// + /// Registers the default client bound from a configuration section. + /// + public void AddDefaultClient(IConfiguration configuration) { - Registry.DefaultClientSettings = new QueueClientSettings(); - settings(Registry.DefaultClientSettings); + ArgumentNullException.ThrowIfNull(configuration); + AddDefaultClient(settings => configuration.Bind(settings)); } -} \ No newline at end of file +} diff --git a/src/AzureStorage.QueueService/QueueClientSettingsRegistry.cs b/src/AzureStorage.QueueService/QueueClientSettingsRegistry.cs index a917f37..127263d 100644 --- a/src/AzureStorage.QueueService/QueueClientSettingsRegistry.cs +++ b/src/AzureStorage.QueueService/QueueClientSettingsRegistry.cs @@ -1,8 +1,18 @@ namespace AzureStorage.QueueService; +/// +/// Records which client names were registered. The settings themselves live in the named-options system +/// (IOptionsMonitor<QueueClientSettings>), keyed by the same names. +/// internal sealed class QueueClientSettingsRegistry { - public Dictionary NamedClientsSettings { get; } = new(); + private readonly HashSet _clientNames = new(StringComparer.Ordinal); - public QueueClientSettings DefaultClientSettings { get; set; } = new(); -} \ No newline at end of file + public bool HasDefaultClient { get; set; } + + public IReadOnlyCollection ClientNames => _clientNames; + + public void Add(string clientName) => _clientNames.Add(clientName); + + public bool Contains(string clientName) => _clientNames.Contains(clientName); +} diff --git a/src/AzureStorage.QueueService/ServiceCollectionExtensions.cs b/src/AzureStorage.QueueService/ServiceCollectionExtensions.cs index ca481bb..22478c3 100644 --- a/src/AzureStorage.QueueService/ServiceCollectionExtensions.cs +++ b/src/AzureStorage.QueueService/ServiceCollectionExtensions.cs @@ -1,22 +1,36 @@ using System.Text.Json; using AzureStorage.QueueService.Telemetry; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; namespace AzureStorage.QueueService; public static class ServiceCollectionExtensions { - private static readonly QueueClientSettingsRegistry Registry = new(); - - public static IServiceCollection AddAzureStorageQueueClient(this IServiceCollection services, Action azureStorageQueueClientBuilderDelegate, JsonSerializerOptions? serializerOptions = null) + /// + /// Registers the queue client factory and the clients configured in . + /// Can be called more than once; later calls add clients to the same registration. + /// + /// + /// Serializer options shared by every client that does not set its own . + /// When omitted, an you registered yourself is used if present. + /// + public static IServiceCollection AddAzureStorageQueueClient(this IServiceCollection services, Action configure, JsonSerializerOptions? serializerOptions = null) { - var builder = new QueueClientSettingsBuilder(Registry); - azureStorageQueueClientBuilderDelegate(builder); + ArgumentNullException.ThrowIfNull(services); + ArgumentNullException.ThrowIfNull(configure); + + var registry = GetOrAddRegistry(services); + + services.TryAddSingleton(); + services.TryAddSingleton(); + + if (serializerOptions is not null) + services.Replace(ServiceDescriptor.Singleton(new JsonQueueMessageConverter(serializerOptions))); + else + services.TryAddSingleton(new JsonQueueMessageConverter()); - services.AddSingleton(Registry); - services.AddSingleton(); - services.AddSingleton(); - services.AddSingleton(new JsonQueueMessageConverter(serializerOptions)); + configure(new QueueClientSettingsBuilder(services, registry)); return services; } @@ -26,4 +40,19 @@ public static IServiceCollection ConfigureQueueServiceTelemetry(this IServiceCol services.Configure(options); return services; } -} \ No newline at end of file + + // One registry per service collection. It used to be a process-wide static, so every host built in the same + // process (a second host, a test host) silently shared and overwrote one set of client settings. + private static QueueClientSettingsRegistry GetOrAddRegistry(IServiceCollection services) + { + foreach (var descriptor in services) + { + if (descriptor.ServiceType == typeof(QueueClientSettingsRegistry) && descriptor.ImplementationInstance is QueueClientSettingsRegistry existing) + return existing; + } + + var registry = new QueueClientSettingsRegistry(); + services.AddSingleton(registry); + return registry; + } +} diff --git a/src/AzureStorage.QueueService/Telemetry/MetricExtensions.cs b/src/AzureStorage.QueueService/Telemetry/MetricExtensions.cs index 95d5cba..f5b6880 100644 --- a/src/AzureStorage.QueueService/Telemetry/MetricExtensions.cs +++ b/src/AzureStorage.QueueService/Telemetry/MetricExtensions.cs @@ -36,10 +36,9 @@ public static async Task ProcessWithMetricsAsync( } finally { - var elapsedSeconds = Stopwatch.GetElapsedTime(start).TotalSeconds; - histogram.Record(elapsedSeconds, tags); - + // TagList is a struct, so the tag has to be added before the record call to have any effect. tags.Add(QueueServiceDiagnostics.Names.QueueProcessSuccess, success); + histogram.Record(Stopwatch.GetElapsedTime(start).TotalSeconds, tags); } } } diff --git a/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs b/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs index 99d0a75..828680e 100644 --- a/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs +++ b/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs @@ -6,7 +6,7 @@ namespace AzureStorage.QueueService.Telemetry; internal class QueueServiceDiagnostics { public static string ServiceName { get; } = "AzureStorage.QueueService"; - + public static ActivitySource ActivitySource { get; } = new ActivitySource(ServiceName); public static Meter Meter { get; } = new Meter(ServiceName); @@ -64,9 +64,11 @@ public struct Metrics ); } + // Event names only. An ActivityEvent captures its timestamp when constructed, so a shared static instance + // would stamp every event with the time the process started. public struct Events - { - public static ActivityEvent QueueMessageReceived = new ActivityEvent("queue.message_received"); - public static ActivityEvent QueueMessageProcessed = new ActivityEvent("queue.message_processed"); + { + public const string QueueMessageReceived = "queue.message_received"; + public const string QueueMessageProcessed = "queue.message_processed"; } } diff --git a/tests/AzureStorage.QueueClient.IntegrationTests/AzureStorage.QueueClient.IntegrationTests.csproj b/tests/AzureStorage.QueueClient.IntegrationTests/AzureStorage.QueueClient.IntegrationTests.csproj index d92f80d..f03f179 100644 --- a/tests/AzureStorage.QueueClient.IntegrationTests/AzureStorage.QueueClient.IntegrationTests.csproj +++ b/tests/AzureStorage.QueueClient.IntegrationTests/AzureStorage.QueueClient.IntegrationTests.csproj @@ -10,12 +10,12 @@ - - - - + + + + - + diff --git a/tests/AzureStorage.QueueService.Tests/AzureStorage.QueueService.Tests.csproj b/tests/AzureStorage.QueueService.Tests/AzureStorage.QueueService.Tests.csproj index d864feb..23316d8 100644 --- a/tests/AzureStorage.QueueService.Tests/AzureStorage.QueueService.Tests.csproj +++ b/tests/AzureStorage.QueueService.Tests/AzureStorage.QueueService.Tests.csproj @@ -9,18 +9,17 @@ - - - - - - + + + + + - + runtime; build; native; contentfiles; analyzers; buildtransitive all - + runtime; build; native; contentfiles; analyzers; buildtransitive all diff --git a/tests/AzureStorage.QueueService.Tests/AzureStorageQueueServiceTests.cs b/tests/AzureStorage.QueueService.Tests/AzureStorageQueueServiceTests.cs index 3150de4..3bfa41b 100644 --- a/tests/AzureStorage.QueueService.Tests/AzureStorageQueueServiceTests.cs +++ b/tests/AzureStorage.QueueService.Tests/AzureStorageQueueServiceTests.cs @@ -1,11 +1,8 @@ -using AutoFixture; -using Azure; -using Azure.Messaging; +using Azure; using Azure.Storage.Queues; using Azure.Storage.Queues.Models; using AzureStorage.QueueService.Telemetry; using FluentAssertions; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -13,144 +10,208 @@ namespace AzureStorage.QueueService.Tests; -public class AzureStorageQueueServiceTests : BaseTestHost +public class AzureStorageQueueServiceTests { - private readonly QueueClientSettings _queueClientSettings; - private readonly IOptions _telemetrySettings = Options.Create(new QueueServiceTelemetrySettings()); - private readonly IServiceProvider _serviceProvider; + private static readonly IOptions TelemetrySettings = Options.Create(new QueueServiceTelemetrySettings()); + private static readonly ILoggerFactory LoggerFactory = new ServiceCollection().AddLogging().BuildServiceProvider().GetRequiredService(); + private const string ValidJson = "{\"Property1\":\"test\",\"Property2\":true,\"Property3\":1}"; - public AzureStorageQueueServiceTests() + private static Mock QueueClientReturning(params string[] messageIds) { - var services = new ServiceCollection(); - services.AddLogging(); - - _serviceProvider = services.BuildServiceProvider(); - - _queueClientSettings = new(); - Configuration.Bind(nameof(QueueClientSettings), _queueClientSettings); - - var data = new TestObject - { - Property1 = "test", - Property2 = true, - Property3 = 1 - }; - var cloudEvent = new CloudEvent("source", "myClass", data, typeof(TestObject)) - { - Id = "myCustomId", - }; + var messages = messageIds.Select(id => QueuesModelFactory.QueueMessage(id, $"receipt-{id}", ValidJson, 1)).ToArray(); + var mock = new Mock(); + mock.Setup(x => x.ReceiveMessagesAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Response.FromValue(messages, Mock.Of())); + return mock; } + private static AzureStorageQueueClient CreateSubject(Mock queueClient, TimeSpan? visibilityTimeout = null) => + new(new JsonQueueMessageConverter(), queueClient.Object, LoggerFactory, TelemetrySettings, visibilityTimeout); + + private static ValueTask Ignore(Exception exception, IDictionary? metadata) => ValueTask.CompletedTask; + [Fact(DisplayName = "Peek messages returns message collection")] public async Task Peek_Messages_Returns_Collection() { // arrange - var fixture = new Fixture(); - Response mockResponse = Mock.Of(); - var peekedMessage = QueuesModelFactory.PeekedMessage("1", "test_text", 1); - PeekedMessage[] peekedMessages = { peekedMessage }; - var response = Response.FromValue(peekedMessages, mockResponse); - + var peekedMessages = new[] { QueuesModelFactory.PeekedMessage("1", ValidJson, 1) }; var mockQueueClient = new Mock(); - mockQueueClient.Setup(x => x.PeekMessagesAsync(It.IsAny(), CancellationToken.None)).ReturnsAsync(response); + mockQueueClient.Setup(x => x.PeekMessagesAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(Response.FromValue(peekedMessages, Mock.Of())); - var mockMessageConverter = new Mock(); - mockMessageConverter.Setup(x => x.Convert(It.IsAny())).Returns(fixture.Create()); + // act + var messages = await CreateSubject(mockQueueClient).PeekMessagesAsync(1); - var loggerFactory = _serviceProvider.GetRequiredService(); + // assert + messages.Should().ContainSingle().Which.Property1.Should().Be("test"); + } - var subject = new AzureStorageQueueClient(mockMessageConverter.Object, mockQueueClient.Object, loggerFactory, _telemetrySettings); + [Fact(DisplayName = "Receive handles every message in the batch and deletes each one")] + public async Task Receive_Handles_And_Deletes_Each_Message() + { + // arrange + var mockQueueClient = QueueClientReturning("1", "2"); + var handled = new List(); // act - var messages = await subject.PeekMessagesAsync(It.IsAny()); + await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (message, _) => { handled.Add(message); return ValueTask.CompletedTask; }, + Ignore, + numMessages: 2); // assert - messages.Should().NotBeEmpty(); - messages.Count().Should().Be(1); + handled.Should().HaveCount(2); + mockQueueClient.Verify(x => x.DeleteMessageAsync("1", "receipt-1", It.IsAny()), Times.Once); + mockQueueClient.Verify(x => x.DeleteMessageAsync("2", "receipt-2", It.IsAny()), Times.Once); } - [Fact(DisplayName = "Peek messages returns message collection")] - public async Task Receive_Messages_Returns_Collection() + [Fact(DisplayName = "A throwing handler leaves its message on the queue and the rest of the batch is still processed")] + public async Task Receive_When_Handler_Throws_Leaves_Message_And_Continues() { // arrange - var fixture = new Fixture(); - Response mockResponse = Mock.Of(); - var queueMessage = QueuesModelFactory.QueueMessage("1", "2", "test_text", 1); - QueueMessage[] peekedMessages = { queueMessage }; - var response = Response.FromValue(peekedMessages, mockResponse); + var mockQueueClient = QueueClientReturning("1", "2"); + var reported = new List(); + var calls = 0; + + // act + await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (_, _) => ++calls == 1 ? throw new InvalidOperationException("Hello from Exception") : ValueTask.CompletedTask, + (exception, _) => { reported.Add(exception); return ValueTask.CompletedTask; }, + numMessages: 2); + // assert + calls.Should().Be(2); + reported.Should().ContainSingle().Which.Message.Should().Be("Hello from Exception"); + mockQueueClient.Verify(x => x.DeleteMessageAsync("1", It.IsAny(), It.IsAny()), Times.Never); + mockQueueClient.Verify(x => x.DeleteMessageAsync("2", It.IsAny(), It.IsAny()), Times.Once); + } + + [Fact(DisplayName = "A failed receive is reported to the exception handler instead of thrown")] + public async Task Receive_When_Receive_Fails_Reports_Instead_Of_Throwing() + { + // arrange var mockQueueClient = new Mock(); - mockQueueClient.Setup(x => x.ReceiveMessagesAsync(1, null, CancellationToken.None)).ReturnsAsync(response); - mockQueueClient.Setup(x => x.DeleteMessageAsync(It.IsAny(), It.IsAny(), CancellationToken.None)); + mockQueueClient.Setup(x => x.ReceiveMessagesAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(500, "storage is having a bad day")); + var reported = new List(); - var mockMessageConverter = new Mock(); - mockMessageConverter.Setup(x => x.Convert(It.IsAny())).Returns(fixture.Create()); + // act + Func act = async () => await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (_, _) => throw new Exception("handler should not run"), + (exception, _) => { reported.Add(exception); return ValueTask.CompletedTask; }); - var loggerFactory = _serviceProvider.GetRequiredService(); + // assert + await act.Should().NotThrowAsync(); + reported.Should().ContainSingle().Which.Should().BeOfType(); + } - var subject = new AzureStorageQueueClient(mockMessageConverter.Object, mockQueueClient.Object, loggerFactory, _telemetrySettings); + [Fact(DisplayName = "A throwing exception handler does not escape and the rest of the batch is still processed")] + public async Task Receive_When_Exception_Handler_Throws_Does_Not_Escape() + { + // arrange + var mockQueueClient = QueueClientReturning("1", "2"); + var handlerCalls = 0; - // act/assert - await subject.ReceiveMessagesAsync(HandleMessage, HandleException); + // act + Func act = async () => await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (_, _) => { handlerCalls++; throw new InvalidOperationException("handler failed"); }, + (_, _) => throw new InvalidOperationException("exception handler failed too"), + numMessages: 2); - ValueTask HandleMessage(TestObject? testObject, IDictionary? metadata) => ValueTask.CompletedTask; - ValueTask HandleException(Exception exception, IDictionary? metadata) => ValueTask.CompletedTask; + // assert + await act.Should().NotThrowAsync(); + handlerCalls.Should().Be(2); } - [Fact(DisplayName = "Peek messages returns message collection")] - public async Task Receive_Messages_Throws_Collection() + [Fact(DisplayName = "A delete that fails after the handler succeeded is reported to the exception handler")] + public async Task Receive_When_Delete_Fails_Reports_To_Exception_Handler() { // arrange - var fixture = new Fixture(); - var mockResponse = Mock.Of(); - var queueMessage = QueuesModelFactory.QueueMessage("1", "2", "test_text", 1); - QueueMessage[] peekedMessages = { queueMessage }; - var response = Response.FromValue(peekedMessages, mockResponse); - - var mockQueueClient = new Mock(_queueClientSettings.ConnectionString!, _queueClientSettings.QueueName!); - mockQueueClient.Setup(x => x.ReceiveMessagesAsync(1, null, CancellationToken.None)).ReturnsAsync(response); - mockQueueClient.Setup(x => x.DeleteMessageAsync(It.IsAny(), It.IsAny(), CancellationToken.None)); - - var mockMessageConverter = new Mock(); - mockMessageConverter.Setup(x => x.Convert(It.IsAny())).Returns(fixture.Create()); - - var loggerFactory = _serviceProvider.GetRequiredService(); - - var subject = new AzureStorageQueueClient(mockMessageConverter.Object, mockQueueClient.Object, loggerFactory, _telemetrySettings); - - // act/assert - await subject.ReceiveMessagesAsync( - (message, metadata) => throw new Exception("Hello from Exception"), - (exception, metadata) => - { - exception.Message.Should().Be("Hello from Exception"); - return ValueTask.CompletedTask; - }); + var mockQueueClient = QueueClientReturning("1"); + mockQueueClient.Setup(x => x.DeleteMessageAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "pop receipt mismatch")); + var reported = new List(); + + // act + await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (_, _) => ValueTask.CompletedTask, + (exception, _) => { reported.Add(exception); return ValueTask.CompletedTask; }); + + // assert + reported.Should().ContainSingle().Which.Should().BeOfType(); } - [Fact(DisplayName = "Can send message")] - public async Task Can_Send_Message() + [Fact(DisplayName = "A message that cannot be deserialized is reported and left on the queue")] + public async Task Receive_When_Conversion_Fails_Reports_And_Leaves_Message() + { + // arrange + var mockQueueClient = new Mock(); + mockQueueClient.Setup(x => x.ReceiveMessagesAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(Response.FromValue(new[] { QueuesModelFactory.QueueMessage("1", "receipt-1", "this is not json", 3) }, Mock.Of())); + var reported = new List(); + + // act + await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (_, _) => throw new Exception("handler should not run"), + (exception, _) => { reported.Add(exception); return ValueTask.CompletedTask; }); + + // assert + reported.Should().ContainSingle().Which.Should().BeOfType() + .Which.InnerException.Should().NotBeNull(); + mockQueueClient.Verify(x => x.DeleteMessageAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact(DisplayName = "The caller's own cancellation still propagates")] + public async Task Receive_When_Caller_Cancels_Throws_OperationCanceled() { // arrange - var fixture = new Fixture(); - var testObject = fixture.Create(); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + var mockQueueClient = new Mock(); + mockQueueClient.Setup(x => x.ReceiveMessagesAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new OperationCanceledException(cts.Token)); + var exceptionHandlerCalled = false; + + // act + Func act = async () => await CreateSubject(mockQueueClient).ReceiveMessagesAsync( + (_, _) => ValueTask.CompletedTask, + (_, _) => { exceptionHandlerCalled = true; return ValueTask.CompletedTask; }, + cancellationToken: cts.Token); + + // assert + await act.Should().ThrowAsync(); + exceptionHandlerCalled.Should().BeFalse(); + } - var mockQueueClient = new Mock(_queueClientSettings.ConnectionString!, _queueClientSettings.QueueName!); - var mockResponse = Mock.Of(); - var mockMessageConverter = new Mock(); + [Fact(DisplayName = "The configured visibility timeout is passed to the receive call")] + public async Task Receive_Passes_Configured_Visibility_Timeout() + { + // arrange + var mockQueueClient = QueueClientReturning(); + var visibilityTimeout = TimeSpan.FromMinutes(5); - mockMessageConverter.Setup(x => x.Convert(testObject)).Returns(It.IsAny()); + // act + await CreateSubject(mockQueueClient, visibilityTimeout).ReceiveMessagesAsync((_, _) => ValueTask.CompletedTask, Ignore); - SendReceipt sendReceipt = - QueuesModelFactory.SendReceipt("1", DateTimeOffset.Now, DateTimeOffset.Now, "2", DateTimeOffset.Now); - var response = Response.FromValue(sendReceipt, mockResponse); - mockQueueClient.Setup(x => x.SendMessageAsync(It.IsAny(), null, null, CancellationToken.None)).ReturnsAsync(response); + // assert + mockQueueClient.Verify(x => x.ReceiveMessagesAsync(1, visibilityTimeout, It.IsAny()), Times.Once); + } - var loggerFactory = _serviceProvider.GetRequiredService(); + [Fact(DisplayName = "Can send message")] + public async Task Can_Send_Message() + { + // arrange + var testObject = new TestObject { Property1 = "sent", Property2 = true, Property3 = 1 }; + var sendReceipt = QueuesModelFactory.SendReceipt("1", DateTimeOffset.Now, DateTimeOffset.Now, "2", DateTimeOffset.Now); + var mockQueueClient = new Mock(); + mockQueueClient.Setup(x => x.SendMessageAsync(It.IsAny(), null, null, It.IsAny())) + .ReturnsAsync(Response.FromValue(sendReceipt, Mock.Of())); - var subject = new AzureStorageQueueClient(mockMessageConverter.Object, mockQueueClient.Object, loggerFactory, _telemetrySettings); + // act + var receipt = await CreateSubject(mockQueueClient).SendMessageAsync(testObject); - // act/assert - await subject.Invoking(async x => await x.SendMessageAsync(testObject)).Should().NotThrowAsync(); + // assert + receipt.Should().BeSameAs(sendReceipt); + mockQueueClient.Verify(x => x.SendMessageAsync(It.Is(b => b.ToString().Contains("\"Property1\":\"sent\"")), null, null, It.IsAny()), Times.Once); } -} \ No newline at end of file +} diff --git a/tests/AzureStorage.QueueService.Tests/JsonQueueMessageConverterTests.cs b/tests/AzureStorage.QueueService.Tests/JsonQueueMessageConverterTests.cs index 2dde07d..33c0c0f 100644 --- a/tests/AzureStorage.QueueService.Tests/JsonQueueMessageConverterTests.cs +++ b/tests/AzureStorage.QueueService.Tests/JsonQueueMessageConverterTests.cs @@ -1,6 +1,5 @@ using System.Text; using System.Text.Json; -using AutoFixture; using Azure.Storage.Queues.Models; using FluentAssertions; @@ -8,12 +7,13 @@ namespace AzureStorage.QueueService.Tests; public class JsonQueueMessageConverterTests { + private static TestObject NewTestObject() => new() { Property1 = Guid.NewGuid().ToString(), Property2 = true, Property3 = Random.Shared.Next() }; + [Fact(DisplayName = "Convert into object")] public void Convert_To_Object() { // arrange - var fixture = new Fixture(); - var testObject = fixture.Create(); + var testObject = NewTestObject(); var json = JsonSerializer.Serialize(testObject); var testObjectBytes = Encoding.UTF8.GetBytes(json); var queueMessage = QueuesModelFactory.QueueMessage("1", "2", new BinaryData(testObjectBytes), 1); @@ -26,38 +26,49 @@ public void Convert_To_Object() testObject.Should().BeEquivalentTo(newTestObject); } - [Fact(DisplayName = "Convert empty string returns null")] - public void Convert_Empty_Binary_Returns_Null() + [Fact(DisplayName = "Convert empty string throws with the serializer error attached")] + public void Convert_Empty_String_Throws() { // arrange var subject = new JsonQueueMessageConverter(); // act/assert - subject.Invoking(x => x.Convert(string.Empty)).Should().Throw(); + subject.Invoking(x => x.Convert(string.Empty)).Should().Throw() + .WithInnerException(); } - [Fact(DisplayName = "Convert object to BinaryData")] - public void Convert_Object_To_BinaryData() + [Fact(DisplayName = "Convert Base64-encoded JSON (as written by older SDKs and Azure Functions)")] + public void Convert_Base64_Encoded_Json() { // arrange - var fixture = new Fixture(); - var testObject = fixture.Create(); - + var testObject = NewTestObject(); + var encoded = Convert.ToBase64String(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(testObject))); var subject = new JsonQueueMessageConverter(); // act - BinaryData binaryData = subject.Convert(testObject); + var result = subject.Convert(encoded); // assert - binaryData.Should().BeOfType(typeof(BinaryData)); + result.Should().BeEquivalentTo(testObject); + } + + [Theory(DisplayName = "Short JSON values that are also valid Base64 are read as JSON")] + [InlineData("1234", 1234)] + [InlineData("true", true)] + public void Convert_Short_Json_That_Looks_Like_Base64_Is_Not_Mangled(string input, T expected) + { + // arrange + var subject = new JsonQueueMessageConverter(); + + // act/assert + subject.Convert(input).Should().Be(expected); } [Fact(DisplayName = "Convert from Object to BinaryData and back to Object")] public void Convert_To_And_From() { // arrange - var fixture = new Fixture(); - var testObject = fixture.Create(); + var testObject = NewTestObject(); var subject = new JsonQueueMessageConverter(); @@ -68,4 +79,20 @@ public void Convert_To_And_From() // assert testObject.Should().BeEquivalentTo(convertedTestObject); } -} \ No newline at end of file + + [Fact(DisplayName = "Custom serializer options are applied when sending as well as receiving")] + public void Convert_Uses_Serializer_Options_When_Sending() + { + // arrange + var testObject = new TestObject { Property1 = "value", Property2 = true, Property3 = 1 }; + var subject = new JsonQueueMessageConverter(new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }); + + // act + var json = subject.Convert(testObject).ToString(); + var roundTripped = subject.Convert(json); + + // assert + json.Should().Contain("\"property1\"").And.NotContain("\"Property1\""); + roundTripped.Should().BeEquivalentTo(testObject); + } +} diff --git a/tests/AzureStorage.QueueService.Tests/QueueClientBuilderTests.cs b/tests/AzureStorage.QueueService.Tests/QueueClientBuilderTests.cs index 20d68a5..b681644 100644 --- a/tests/AzureStorage.QueueService.Tests/QueueClientBuilderTests.cs +++ b/tests/AzureStorage.QueueService.Tests/QueueClientBuilderTests.cs @@ -5,23 +5,42 @@ namespace AzureStorage.QueueService.Tests; public class QueueClientBuilderTests { + private const string Description = "Queue client 'test'"; + [Fact] - public void CreateClient_Returns_QueueClient() + public void CreateClient_WithEndpointAndCredential_ReturnsQueueClient() { // arrange var settings = new QueueClientSettings() { TokenCredential = new DefaultAzureCredential(), - EndpointUri = new Uri("https://fake.com"), + EndpointUri = new Uri("https://fake.com/queue1"), + }; + var subject = new QueueClientBuilder(); + + // act + var client = subject.CreateClient(Description, settings); + + // assert + client.Name.Should().Be("queue1"); + } + + [Fact] + public void CreateClient_WithConnectionString_ReturnsQueueClient() + { + // arrange + var settings = new QueueClientSettings() + { + ConnectionString = "UseDevelopmentStorage=true", QueueName = "queue1", }; var subject = new QueueClientBuilder(); // act - var client = subject.CreateClient(settings); + var client = subject.CreateClient(Description, settings); // assert - client.Should().NotBeNull(); + client.Name.Should().Be("queue1"); } [Fact] @@ -36,11 +55,25 @@ public void CreateClient_WithoutUri_Throws() var subject = new QueueClientBuilder(); // act & assert - subject.Invoking(x => x.CreateClient(settings)).Should().Throw(); + subject.Invoking(x => x.CreateClient(Description, settings)).Should().Throw().WithMessage($"{Description}*"); + } + + [Fact] + public void CreateClient_WithConnectionStringButNoQueueName_Throws() + { + // arrange + var settings = new QueueClientSettings() + { + ConnectionString = "UseDevelopmentStorage=true", + }; + var subject = new QueueClientBuilder(); + + // act & assert + subject.Invoking(x => x.CreateClient(Description, settings)).Should().Throw().WithMessage("*QueueName*"); } [Fact] - public void CreateClient_WithoutConnectionString_Throws() + public void CreateClient_WithNothingConfigured_Throws() { // arrange var settings = new QueueClientSettings() @@ -50,6 +83,6 @@ public void CreateClient_WithoutConnectionString_Throws() var subject = new QueueClientBuilder(); // act & assert - subject.Invoking(x => x.CreateClient(settings)).Should().Throw(); + subject.Invoking(x => x.CreateClient(Description, settings)).Should().Throw(); } -} \ No newline at end of file +} diff --git a/tests/AzureStorage.QueueService.Tests/QueueClientFactoryTests.cs b/tests/AzureStorage.QueueService.Tests/QueueClientFactoryTests.cs index 610c31b..b370332 100644 --- a/tests/AzureStorage.QueueService.Tests/QueueClientFactoryTests.cs +++ b/tests/AzureStorage.QueueService.Tests/QueueClientFactoryTests.cs @@ -1,205 +1,192 @@ -using Azure.Identity; -using Azure.Storage.Queues; -using AzureStorage.QueueService.Telemetry; -using FakeItEasy; +using Azure.Storage.Queues; using FluentAssertions; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; using Moq; namespace AzureStorage.QueueService.Tests; -public class QueueClientFactoryTests : BaseTestHost +public class QueueClientFactoryTests { - private readonly IOptions _telemetrySettings = Options.Create(new QueueServiceTelemetrySettings()); + private const string ConnectionString = "UseDevelopmentStorage=true"; - [Fact] - public void QueueClientFactory_GetNamedClient_ReturnsClient() + /// + /// A service collection whose SDK client construction is stubbed out so no settings validation or network + /// access happens. Every call to the builder is recorded in created. + /// + private static (IServiceCollection Services, List<(string Description, QueueClientSettings Settings)> Created) CreateServices() { - // arrange + var created = new List<(string, QueueClientSettings)>(); + var builder = new Mock(); + builder.Setup(x => x.CreateClient(It.IsAny(), It.IsAny())) + .Returns((string description, QueueClientSettings settings) => + { + created.Add((description, settings)); + return new Mock().Object; + }); + var services = new ServiceCollection(); services.AddLogging(); - services.AddAzureStorageQueueClient(options => options.AddClient("foo", y => Configuration.Bind("QueueClientSettings", y))); + services.AddSingleton(builder.Object); // registered first so the library's TryAdd keeps it - var serviceProvider = services.BuildServiceProvider(); + return (services, created); + } - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + [Fact] + public void GetNamedClient_ReturnsSameClientEachTime() + { + // arrange + var (services, created) = CreateServices(); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", y => { y.ConnectionString = ConnectionString; y.QueueName = "foo-queue"; })); + var factory = services.BuildServiceProvider().GetRequiredService(); // act - var queueClient = subject.GetQueueClient("foo"); + var client1 = factory.GetQueueClient("foo"); + var client2 = factory.GetQueueClient("foo"); // assert - queueClient.Should().NotBeNull(); + client1.Should().BeSameAs(client2); + created.Should().ContainSingle().Which.Settings.QueueName.Should().Be("foo-queue"); } [Fact] - public void QueueClientFactory_AddStorageQueueCalledTwice_ReturnsClient() + public void RegisterMultipleNamedClients_EachGetsItsOwnSettings() { // arrange - var services = new ServiceCollection(); - services.AddLogging(); - services.AddAzureStorageQueueClient(options => options.AddClient("foo", y => Configuration.Bind("QueueClientSettings", y))); - services.AddAzureStorageQueueClient(options => options.AddClient("foo", y => Configuration.Bind("QueueClientSettings", y))); - - var serviceProvider = services.BuildServiceProvider(); - - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + var (services, created) = CreateServices(); + services.AddAzureStorageQueueClient(x => + { + x.AddClient("foo", y => { y.ConnectionString = ConnectionString; y.QueueName = "foo-queue"; }); + x.AddClient("bar", y => { y.ConnectionString = ConnectionString; y.QueueName = "bar-queue"; }); + }); + var factory = services.BuildServiceProvider().GetRequiredService(); // act - var queueClient = subject.GetQueueClient("foo"); + var foo = factory.GetQueueClient("foo"); + var bar = factory.GetQueueClient("bar"); // assert - queueClient.Should().NotBeNull(); + foo.Should().NotBeSameAs(bar); + created.Select(c => c.Settings.QueueName).Should().BeEquivalentTo("foo-queue", "bar-queue"); } [Fact] - public void GetNamedClientTwice_ReturnsSameClient() + public void AddClientTwiceForSameName_ComposesConfiguration() { - // arrange - var services = new ServiceCollection(); - services.AddLogging(); - services.AddAzureStorageQueueClient(options => options.AddClient("foo", y => Configuration.Bind("QueueClientSettings", y))); - - var serviceProvider = services.BuildServiceProvider(); - - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + // arrange + var (services, created) = CreateServices(); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", y => y.ConnectionString = ConnectionString)); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", y => y.QueueName = "foo-queue")); + var factory = services.BuildServiceProvider().GetRequiredService(); // act - var queueClient1 = subject.GetQueueClient("foo"); - var queueClient2 = subject.GetQueueClient("foo"); + factory.GetQueueClient("foo"); // assert - queueClient1.Should().NotBeNull(); - queueClient2.Should().NotBeNull(); - queueClient1.Should().BeSameAs(queueClient2); + var settings = created.Should().ContainSingle().Which.Settings; + settings.ConnectionString.Should().Be(ConnectionString); + settings.QueueName.Should().Be("foo-queue"); } [Fact] - public void RegisterMultipleNamedClient_ReturnsCorrectClient() + public void SeparateServiceCollections_DoNotShareSettings() { // arrange - var services = new ServiceCollection(); - services.AddLogging(); - services.AddAzureStorageQueueClient(options => - { - options.AddClient("foo", y => Configuration.Bind("QueueClientSettings", y)); - options.AddClient("bar", y => Configuration.Bind("QueueClientSettings", y)); - }); - - var serviceProvider = services.BuildServiceProvider(); - - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + var (servicesA, createdA) = CreateServices(); + var (servicesB, createdB) = CreateServices(); + servicesA.AddAzureStorageQueueClient(x => x.AddDefaultClient(y => { y.ConnectionString = ConnectionString; y.QueueName = "queue-a"; })); + servicesB.AddAzureStorageQueueClient(x => x.AddDefaultClient(y => { y.ConnectionString = ConnectionString; y.QueueName = "queue-b"; })); // act - var queueClient1 = subject.GetQueueClient("foo"); - var queueClient2 = subject.GetQueueClient("bar"); + servicesA.BuildServiceProvider().GetRequiredService().GetQueueClient(); + servicesB.BuildServiceProvider().GetRequiredService().GetQueueClient(); // assert - queueClient1.Should().NotBeNull(); - queueClient2.Should().NotBeNull(); + createdA.Should().ContainSingle().Which.Settings.QueueName.Should().Be("queue-a"); + createdB.Should().ContainSingle().Which.Settings.QueueName.Should().Be("queue-b"); } [Fact] - public void GetDefaultClient_ReturnsValidClient() + public void GetUnregisteredClient_ThrowsNamingTheRegisteredClients() { // arrange - var services = new ServiceCollection(); - services.AddLogging(); - services.AddAzureStorageQueueClient(options => options.AddDefaultClient(y => Configuration.Bind("QueueClientSettings", y))); - - var serviceProvider = services.BuildServiceProvider(); + var (services, _) = CreateServices(); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", y => y.ConnectionString = ConnectionString)); + var factory = services.BuildServiceProvider().GetRequiredService(); - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + // act/assert + factory.Invoking(x => x.GetQueueClient("bar")).Should().Throw() + .WithMessage("*'bar'*").WithMessage("*'foo'*"); + } - // act - var queueClient = subject.GetQueueClient(); + [Fact] + public void GetDefaultClient_WhenNoneRegistered_Throws() + { + // arrange + var (services, _) = CreateServices(); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", y => y.ConnectionString = ConnectionString)); + var factory = services.BuildServiceProvider().GetRequiredService(); - // assert - queueClient.Should().NotBeNull(); + // act/assert + factory.Invoking(x => x.GetQueueClient()).Should().Throw().WithMessage("*AddDefaultClient*"); } [Fact] - public void GetUnregisteredClient_Throws() + public void Clients_CanBeInjectedDirectly() { // arrange - var services = new ServiceCollection(); - services.AddLogging(); - services.AddAzureStorageQueueClient(options => options.AddClient("foo", y => Configuration.Bind("QueueClientSettings", y))); - - var serviceProvider = services.BuildServiceProvider(); + var (services, _) = CreateServices(); + services.AddAzureStorageQueueClient(x => + { + x.AddClient("foo", y => { y.ConnectionString = ConnectionString; y.QueueName = "foo-queue"; }); + x.AddDefaultClient(y => { y.ConnectionString = ConnectionString; y.QueueName = "default-queue"; }); + }); + var provider = services.BuildServiceProvider(); + var factory = provider.GetRequiredService(); + + // act/assert + provider.GetRequiredKeyedService("foo").Should().BeSameAs(factory.GetQueueClient("foo")); + provider.GetRequiredKeyedService("foo").Should().BeSameAs(factory.GetQueueClient("foo")); + provider.GetRequiredService().Should().BeSameAs(factory.GetQueueClient()); + provider.GetRequiredService().Should().BeSameAs(factory.GetQueueClient()); + } - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + [Fact] + public void AddClient_FromConfigurationSection_BindsSettings() + { + // arrange + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["Queues:Foo:ConnectionString"] = ConnectionString, + ["Queues:Foo:QueueName"] = "foo-queue", + ["Queues:Foo:VisibilityTimeout"] = "00:05:00", + }).Build(); + var (services, created) = CreateServices(); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", configuration.GetSection("Queues:Foo"))); + var factory = services.BuildServiceProvider().GetRequiredService(); // act - subject.Invoking(x => x.GetQueueClient("bar")).Should().Throw(because: "The client name wasn't registered"); + factory.GetQueueClient("foo"); + + // assert + var settings = created.Should().ContainSingle().Which.Settings; + settings.QueueName.Should().Be("foo-queue"); + settings.VisibilityTimeout.Should().Be(TimeSpan.FromMinutes(5)); } [Fact] - public void CreateClient_WithCorrectSettings_Creates() + public void ConcurrentFirstCalls_CreateOneClient() { // arrange - var services = new ServiceCollection(); - services.AddLogging(); - services.AddAzureStorageQueueClient(options => options.AddDefaultClient(y => - { - y.TokenCredential = new DefaultAzureCredential(); - y.EndpointUri = new Uri("https://fake.com"); - y.QueueName = "foo"; - y.CreateIfNotExists = true; - })); - - var serviceProvider = services.BuildServiceProvider(); - - var registry = serviceProvider.GetRequiredService(); - var loggerFactory = serviceProvider.GetRequiredService(); - var messageConverter = serviceProvider.GetRequiredService(); - var mockQueueClientBuilder = new Mock(); - var mockQueueClient = new Mock(); - mockQueueClientBuilder.Setup(x => x.CreateClient(It.IsAny())).Returns(mockQueueClient.Object); - IQueueClientFactory subject = new QueueClientFactory(registry, mockQueueClientBuilder.Object, loggerFactory, messageConverter, _telemetrySettings); + var (services, created) = CreateServices(); + services.AddAzureStorageQueueClient(x => x.AddClient("foo", y => y.ConnectionString = ConnectionString)); + var factory = services.BuildServiceProvider().GetRequiredService(); // act - var client = subject.GetQueueClient(); + var clients = Enumerable.Range(0, 32).AsParallel().Select(_ => factory.GetQueueClient("foo")).ToList(); // assert - client.Should().NotBeNull(); + created.Should().ContainSingle(); + clients.Distinct().Should().ContainSingle(); } -} \ No newline at end of file +}