Add retry wrapper for ASB message consumer in generated tests

Like Kafka tests, wrap the ASB consumer with RetryAzureServiceBusMessageConsumer
that retries Receive up to 10 times when MT_NONE is returned. This gives CI
environments more time for messages to arrive before tests fail.

Also updates AzureServiceBusMessageGatewayProvider to create channels directly
(injecting the wrapped consumer) instead of using AzureServiceBusChannelFactory.

💘 Generated with Crush

Author: lillo42

tests/Paramore.Brighter.AzureServiceBus.Tests/MessagingGateway/AzureServiceBusMessageGatewayProvider.cs

@@ -84,15 +84,27 @@ public async Task<IAmAMessageProducerAsync> CreateProducerAsync(AzureServiceBusP
     public IAmAChannelSync CreateChannel(AzureServiceBusSubscription subscription)
     {
         var consumerFactory = new AzureServiceBusConsumerFactory(_clientProvider);
-        var channelFactory = new AzureServiceBusChannelFactory(consumerFactory);
-        return channelFactory.CreateSyncChannel(subscription);
+        var consumer = consumerFactory.Create(subscription);
+        var retryConsumer = new RetryAzureServiceBusMessageConsumer(consumer, maxRetries: 10);
+
+        return new Channel(
+            subscription.ChannelName,
+            subscription.RoutingKey,
+            retryConsumer,
+            subscription.BufferSize);
     }
 
     public async Task<IAmAChannelAsync> CreateChannelAsync(AzureServiceBusSubscription subscription, CancellationToken cancellationToken = default)
     {
         var consumerFactory = new AzureServiceBusConsumerFactory(_clientProvider);
-        var channelFactory = new AzureServiceBusChannelFactory(consumerFactory);
-        return await channelFactory.CreateAsyncChannelAsync(subscription, cancellationToken);
+        var consumer = consumerFactory.Create(subscription);
+        var retryConsumer = new RetryAzureServiceBusMessageConsumer(consumer, maxRetries: 10);
+
+        return new ChannelAsync(
+            subscription.ChannelName,
+            subscription.RoutingKey,
+            retryConsumer,
+            subscription.BufferSize);
     }
 
     public void CleanUp(IAmAMessageProducerSync? producer, IAmAChannelSync? channel, IEnumerable<Message> messages)

tests/Paramore.Brighter.AzureServiceBus.Tests/MessagingGateway/RetryAzureServiceBusMessageConsumer.cs

@@ -0,0 +1,117 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Paramore.Brighter.AzureServiceBus.Tests.MessagingGateway;
+
+/// <summary>
+/// A decorator for Azure Service Bus message consumers that retries receiving a message
+/// when the underlying consumer returns an empty result or MT_NONE message.
+/// This is useful in CI environments where ASB may be slow to deliver messages.
+/// </summary>
+public class RetryAzureServiceBusMessageConsumer : IAmAMessageConsumerSync, IAmAMessageConsumerAsync
+{
+    private readonly IAmAMessageConsumerSync _innerSync;
+    private readonly IAmAMessageConsumerAsync _innerAsync;
+    private readonly int _maxRetries;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RetryAzureServiceBusMessageConsumer"/> class.
+    /// </summary>
+    /// <param name="inner">The underlying Azure Service Bus message consumer to decorate.</param>
+    /// <param name="maxRetries">The maximum number of receive attempts (minimum 1).</param>
+    public RetryAzureServiceBusMessageConsumer(IAmAMessageConsumerSync inner, int maxRetries = 10)
+    {
+        _innerSync = inner ?? throw new ArgumentNullException(nameof(inner));
+        _innerAsync = inner as IAmAMessageConsumerAsync ?? throw new ArgumentException("Inner consumer must implement IAmAMessageConsumerAsync", nameof(inner));
+        _maxRetries = Math.Max(1, maxRetries);
+    }
+
+    public void Acknowledge(Message message)
+    {
+        _innerSync.Acknowledge(message);
+    }
+
+    public Task AcknowledgeAsync(Message message, CancellationToken cancellationToken = default)
+    {
+        return _innerAsync.AcknowledgeAsync(message, cancellationToken);
+    }
+
+    public void Purge()
+    {
+        _innerSync.Purge();
+    }
+
+    public Task PurgeAsync(CancellationToken cancellationToken = default)
+    {
+        return _innerAsync.PurgeAsync(cancellationToken);
+    }
+
+    public Message[] Receive(TimeSpan? timeOut = null)
+    {
+        for (var i = 0; i < _maxRetries; i++)
+        {
+            var messages = _innerSync.Receive(timeOut);
+            if (messages.Length > 0 && messages[0].Header.MessageType != MessageType.MT_NONE)
+            {
+                return messages;
+            }
+        }
+
+        return [new Message()];
+    }
+
+    public async Task<Message[]> ReceiveAsync(TimeSpan? timeOut = null, CancellationToken cancellationToken = default)
+    {
+        for (var i = 0; i < _maxRetries; i++)
+        {
+            var messages = await _innerAsync.ReceiveAsync(timeOut, cancellationToken);
+            if (messages.Length > 0 && messages[0].Header.MessageType != MessageType.MT_NONE)
+            {
+                return messages;
+            }
+        }
+
+        return [new Message()];
+    }
+
+    public bool Reject(Message message, MessageRejectionReason? reason = null)
+    {
+        return _innerSync.Reject(message, reason);
+    }
+
+    public Task<bool> RejectAsync(Message message, MessageRejectionReason? reason = null, CancellationToken cancellationToken = default)
+    {
+        return _innerAsync.RejectAsync(message, reason, cancellationToken);
+    }
+
+    public void Nack(Message message)
+    {
+        _innerSync.Nack(message);
+    }
+
+    public Task NackAsync(Message message, CancellationToken cancellationToken = default)
+    {
+        return _innerAsync.NackAsync(message, cancellationToken);
+    }
+
+    public bool Requeue(Message message, TimeSpan? delay = null)
+    {
+        return _innerSync.Requeue(message, delay);
+    }
+
+    public Task<bool> RequeueAsync(Message message, TimeSpan? delay = null, CancellationToken cancellationToken = default)
+    {
+        return _innerAsync.RequeueAsync(message, delay, cancellationToken);
+    }
+
+    public void Dispose()
+    {
+        _innerSync.Dispose();
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _innerAsync.DisposeAsync();
+    }
+}

tests/Paramore.Brighter.AzureServiceBus.Tests/test-configuration.json

@@ -11,7 +11,6 @@
     "HasSupportToDelayedMessages": true,
     "HasSupportToValidateBrokerExistence": true,
     "HasSupportToRequeue": true,
-    "ReceiveMessageTimeoutInMilliseconds": 5000,
-    "DelayBetweenReceiveMessageInMilliseconds": 1000
+    "ReceiveMessageTimeoutInMilliseconds": 6000
   }
 }