Merge pull request #403 from Particular/use-isolated-dispatch-for-con…

…trol-message

Use isolated dispatch for control message to prevent attaching to outbox's transaction scope if transport supports transaction scopes

src/NServiceBus.TransactionalSession.Tests/OutboxTransactionalSessionTests.cs

@@ -5,6 +5,7 @@
     using System.Threading.Tasks;
     using Extensibility;
     using Fakes;
+    using NServiceBus.Transport;
     using NUnit.Framework;
 
     [TestFixture]
@@ -135,6 +136,7 @@ public async Task Commit_should_send_control_message_and_store_outbox_data()
             var controlMessage = dispatched.outgoingMessages.UnicastTransportOperations.Single();
             Assert.Multiple(() =>
             {
+                Assert.That(controlMessage.RequiredDispatchConsistency, Is.EqualTo(DispatchConsistency.Isolated));
                 Assert.That(controlMessage.Message.MessageId, Is.EqualTo(session.SessionId));
                 Assert.That(controlMessage.Message.Headers[Headers.ControlMessageHeader], Is.EqualTo(bool.TrueString));
                 Assert.That(controlMessage.Message.Body.IsEmpty, Is.True);

src/NServiceBus.TransactionalSession.Tests/TransactionalSessionTests.cs

@@ -6,6 +6,7 @@
     using Extensibility;
     using Fakes;
     using NUnit.Framework;
+    using Transport;
 
     [TestFixture]
     public class TransactionalSessionTests

src/NServiceBus.TransactionalSession/OutboxTransactionalSession.cs

@@ -71,7 +71,13 @@ async Task DispatchControlMessage(CancellationToken cancellationToken)
                 }
             }
             var message = new OutgoingMessage(SessionId, headers, ReadOnlyMemory<byte>.Empty);
-            var outgoingMessages = new TransportOperations(new TransportTransportOperation(message, new UnicastAddressTag(physicalQueueAddress)));
+            var operation = new TransportTransportOperation(
+                message,
+                new UnicastAddressTag(physicalQueueAddress),
+                null,
+                DispatchConsistency.Isolated // Avoids promoting to distributed tx by not combining transport and persistence when both share same technology
+                );
+            var outgoingMessages = new TransportOperations(operation);
             await dispatcher.Dispatch(outgoingMessages, new TransportTransaction(), cancellationToken).ConfigureAwait(false);
         }