check if model is closed before sending, and try up to 3 times to send before giving up

Author: mookid8000

CHANGELOG.md

@@ -171,6 +171,8 @@
 * Interpret delivery count header value as ASCII text with a number, when the type encountered is `byte[]`
 * Ensure that RabbitMQ's built-in quorum queue header `x-delivery-count` is cleared when a message is dead-lettered
 
+## 9.4.0
+* Screen the `IModel` for fitness before trying to send with it, and perform send operations with up to 3 attempts to hopefully better overcome transient errors
 
 ---
 

Rebus.RabbitMq.Tests/Rebus.RabbitMq.Tests.csproj

@@ -8,17 +8,17 @@
 		<ProjectReference Include="..\Rebus.RabbitMq\Rebus.RabbitMq.csproj" />
 	</ItemGroup>
 	<ItemGroup>
-		<PackageReference Include="microsoft.net.test.sdk" Version="17.9.0" />
+		<PackageReference Include="microsoft.net.test.sdk" Version="17.10.0" />
 		<PackageReference Include="nunit" Version="3.14.0" />
-		<PackageReference Include="NUnit3TestAdapter" Version="4.5.0">
+		<PackageReference Include="NUnit3TestAdapter" Version="4.6.0">
 			<PrivateAssets>all</PrivateAssets>
 			<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
 		</PackageReference>
 		<PackageReference Include="rabbitmq.client" Version="6.8.1" />
 		<PackageReference Include="rebus.fleetmanager" Version="6.2.0" />
 		<PackageReference Include="rebus.serviceprovider" Version="10.0.0" />
 		<PackageReference Include="rebus.tests.contracts" Version="8.2.6" />
-		<PackageReference Include="Testcontainers.RabbitMq" Version="3.8.0" />
+		<PackageReference Include="Testcontainers.RabbitMq" Version="3.9.0" />
 	</ItemGroup>
 	<ItemGroup>
 		<Folder Include="Internals\" />

Rebus.RabbitMq/Internals/ModelExtensions.cs

@@ -7,7 +7,6 @@ static class ModelExtensions
     /// <summary>
     /// Disposes the specific 
     /// </summary>
-    /// <param name="model"></param>
     internal static void SafeDrop(this IModel model)
     {
         if (model == null)

Rebus.RabbitMq/RabbitMq/RabbitMqTransport.cs

@@ -41,6 +41,11 @@ public class RabbitMqTransport : AbstractRebusTransport, IDisposable, IInitializ
     /// </summary>
     const int QueueDoesNotExist = 404;
 
+    /// <summary>
+    /// Defines how many attempts to make at sending outgoung messages before letting exceptions bubble out
+    /// </summary>
+    const int WriterAttempts = 3;
+
     static readonly Encoding HeaderValueEncoding = Encoding.UTF8;
 
     readonly SemaphoreSlim _consumerLock = new(1, 1);
@@ -54,15 +59,15 @@ public class RabbitMqTransport : AbstractRebusTransport, IDisposable, IInitializ
     readonly SemaphoreSlim _subscriptionSemaphore = new(1, 1);
     readonly ConnectionManager _connectionManager;
     readonly ILog _log;
-
+    
     ushort _maxMessagesToPrefetch;
 
     bool _declareExchanges = true;
     bool _declareInputQueue = true;
     bool _bindInputQueue = true;
     bool _publisherConfirmsEnabled = true;
     int _batchSize = 1;
-
+    
     TimeSpan _publisherConfirmsTimeout;
 
     string _consumerTag = null;
@@ -428,24 +433,46 @@ protected override async Task SendOutgoingMessages(IEnumerable<OutgoingTransport
 
         if (!messages.Any()) return;
 
-        var model = _writerPool.Get();
+        var expressMessages = messages.Where(m => m.IsExpress).Select(m => m.Message).ToList();
+        var ordinaryMessages = messages.Where(m => !m.IsExpress).Select(m => m.Message).ToList();
 
-        try
+        var attempt = 0;
+
+        while (true)
         {
-            var expressMessages = messages.Where(m => m.IsExpress).Select(m => m.Message).ToList();
-            var ordinaryMessages = messages.Where(m => !m.IsExpress).Select(m => m.Message).ToList();
+            var model = _writerPool.Get();
+
+            try
+            {
+                // if the model is not fit, discard it and try a new one
+                if (model.IsClosed)
+                {
+                    model.SafeDrop();
+                    continue;
+                }
 
-            DoSend(expressMessages, model, isExpress: true);
+                // otherwise, count this as an attempt and try to do it
+                attempt++;
 
-            DoSend(ordinaryMessages, model, isExpress: false);
+                DoSend(expressMessages, model, isExpress: true);
+                DoSend(ordinaryMessages, model, isExpress: false);
+                
+                // remember to return the model
+                _writerPool.Return(model);
+                
+                return; //< success - we're done!
+            }
+            catch (Exception)
+            {
+                // if anything goes wrong when using this IModel, asummed it's faulted and drop it
+                model.SafeDrop();
 
-            _writerPool.Return(model);
-        }
-        catch (Exception)
-        {
-            // if anything goes wrong when using this IModel, just drop it
-            model.SafeDrop();
-            throw;
+                // if the built-in number of retries has been exceeded, let the error bubble out
+                if (attempt > WriterAttempts)
+                {
+                    throw;
+                }
+            }
         }
     }