Better performance on publisher confirms in transaction

Rebus.RabbitMq.Tests/RabbitMqPublisherConfirmsPublishPerformanceTest.cs

@@ -7,6 +7,7 @@
 using Rebus.Config;
 using Rebus.Logging;
 using Rebus.Tests.Contracts;
+using Rebus.Transport;
 #pragma warning disable 1998
 
 namespace Rebus.RabbitMq.Tests
@@ -19,10 +20,15 @@ public class RabbitMqPublisherConfirmsPublishPerformanceTest : FixtureBase
         /// <summary>
         /// Without confirms: 15508.5 msg/s
         ///
-        /// With confirms
+        /// With confirms without transaction
         ///     - initial:                      645 msg/s
         ///     - move call to ConfirmSelect:   815.0 msg/s
         /// 
+        /// (The following was done on a different machine, not directly comparable to other results)
+        /// With confirms and in transaction
+        ///     - initial                       118 msg/s
+        ///     - moved outside of send loop:   4315 msg/s 
+        /// 
         /// </summary>
         [TestCase(true, 10000)]
         [TestCase(false, 10000)]
@@ -44,23 +50,48 @@ public async Task PublishBunchOfMessages(bool enablePublisherConfirms, int count
                     .EnablePublisherConfirms(value: enablePublisherConfirms))
                 .Start();
 
-            var stopwatch = Stopwatch.StartNew();
+            // In transaction
+            using(var scope = new RebusTransactionScope())
+            {
+                var stopwatch = Stopwatch.StartNew();
+
+                await Task.WhenAll(Enumerable.Range(0, count)
+                    .Select(n => $"THIS IS MESSAGE NUMBER {n} OUT OF {count}")
+                    .Select(str => activator.Bus.SendLocal(str)));
+
+                await scope.CompleteAsync();
+
+                var elapsedSeconds = stopwatch.Elapsed.TotalSeconds;
+
+                Console.WriteLine($@"Publishing 
+
+                    {count} 
+
+                messages in transaction with PUBLISHER CONFIRMS = {enablePublisherConfirms} took 
+
+                    {elapsedSeconds:0.0} s
+
+                - that's {count/elapsedSeconds:0.0} msg/s");
+             }
+
+            // Without transaction
+            var stopwatch2 = Stopwatch.StartNew();
 
             await Task.WhenAll(Enumerable.Range(0, count)
                 .Select(n => $"THIS IS MESSAGE NUMBER {n} OUT OF {count}")
                 .Select(str => activator.Bus.SendLocal(str)));
 
-            var elapsedSeconds = stopwatch.Elapsed.TotalSeconds;
+            var elapsedSeconds2 = stopwatch2.Elapsed.TotalSeconds;         
 
             Console.WriteLine($@"Publishing 
 
-    {count} 
+                {count} 
 
-messages with PUBLISHER CONFIRMS = {enablePublisherConfirms} took 
+            messages without transaction with PUBLISHER CONFIRMS = {enablePublisherConfirms} took 
 
-    {elapsedSeconds:0.0} s
+                {elapsedSeconds2:0.0} s
 
-- that's {count/elapsedSeconds:0.0} msg/s");
+            - that's {count/elapsedSeconds2:0.0} msg/s");
         }
     }
 }

Rebus.RabbitMq/RabbitMq/RabbitMqTransport.cs

@@ -502,6 +502,11 @@ async Task SendOutgoingMessages(ITransactionContext context, ConcurrentQueue<Out
         {
             var model = GetModel(context);
 
+            if (_publisherConfirmsEnabled)
+            {
+                model.ConfirmSelect();
+            }
+
             foreach (var outgoingMessage in outgoingMessages)
             {
                 var destinationAddress = outgoingMessage.DestinationAddress;
@@ -526,23 +531,18 @@ async Task SendOutgoingMessages(ITransactionContext context, ConcurrentQueue<Out
                     EnsureQueueExists(routingKey, model);
                 }
 
-                if (_publisherConfirmsEnabled)
-                {
-                    model.ConfirmSelect();
-                }
-
                 model.BasicPublish(
                     exchange: exchange,
                     routingKey: routingKey.RoutingKey,
                     mandatory: mandatory,
                     basicProperties: props,
                     body: message.Body
                 );
+            }
 
-                if (_publisherConfirmsEnabled)
-                {
-                    model.WaitForConfirmsOrDie();
-                }
+            if (_publisherConfirmsEnabled)
+            {
+                model.WaitForConfirmsOrDie();
             }
         }