Merge pull request #349 from Particular/hotfix-2.2.5

Using raw ReplyTo header when setting the ReplyTo column value

src/NServiceBus.SqlServer.AcceptanceTests/NServiceBus.SqlServer.AcceptanceTests.csproj

@@ -208,6 +208,7 @@
     <Compile Include="When_in_native_transaction_mode.cs" />
     <Compile Include="When_processing_messages.cs" />
     <Compile Include="When_ReplyTo_address_does_not_exist.cs" />
+    <Compile Include="When_sending_raw_transport_messages.cs" />
     <Compile Include="When_using_non_standard_schema.cs" />
     <Compile Include="When_callback_receiver_is_disabled.cs" />
     <Compile Include="When_using_different_connection_strings_for_each_endpoint.cs" />

src/NServiceBus.SqlServer.AcceptanceTests/When_sending_raw_transport_messages.cs

@@ -0,0 +1,83 @@
+namespace NServiceBus.AcceptanceTests.Basic
+{
+    using System;
+    using System.Collections.Generic;
+    using NServiceBus.AcceptanceTesting;
+    using NServiceBus.AcceptanceTesting.Customization;
+    using NServiceBus.AcceptanceTests.EndpointTemplates;
+    using NServiceBus.Satellites;
+    using NServiceBus.Transports;
+    using NServiceBus.Unicast;
+    using NUnit.Framework;
+
+    public class When_sending_raw_transport_messages : NServiceBusAcceptanceTest
+    {
+        static string SenderEndpoint => Conventions.EndpointNamingConvention(typeof(Sender));
+        static string SatelliteAddress => SenderEndpoint + ".Satellite";
+
+        [Test]
+        public void Should_not_modify_reply_to_header()
+        {
+            var context = new Context
+            {
+                Id = Guid.NewGuid()
+            };
+
+            Scenario.Define(context)
+                    .WithEndpoint<Sender>(b => b.Given((bus, ctx) =>
+                    {
+                        var sender = ((UnicastBus) bus).Builder.Build<ISendMessages>();
+                        var headers = new Dictionary<string, string>
+                        {
+                            [Headers.ReplyToAddress] = "ReplyHere@SomeSchema"
+                        };
+                        var message = new TransportMessage(Guid.NewGuid().ToString(), headers);
+                        sender.Send(message, new SendOptions(SatelliteAddress));
+                    }))
+                    .Done(c => c.WasCalled)
+                    .Run();
+
+            Assert.True(context.WasCalled, "The message handler should be called");
+            Assert.AreEqual("ReplyHere@SomeSchema", context.ReplyToAddress);
+        }
+
+
+        public class Context : ScenarioContext
+        {
+            public bool WasCalled { get; set; }
+            public string ReplyToAddress { get; set; }
+            public Guid Id { get; set; }
+        }
+
+        public class Sender : EndpointConfigurationBuilder
+        {
+            public Sender()
+            {
+                EndpointSetup<DefaultServer>();
+            }
+        }
+
+        public class CheckSatellite : ISatellite
+        {
+            public Context Context { get; set; }
+
+            public bool Handle(TransportMessage message)
+            {
+                Context.ReplyToAddress = message.Headers[Headers.ReplyToAddress];
+                Context.WasCalled = true;
+                return true;
+            }
+
+            public void Start()
+            {
+            }
+
+            public void Stop()
+            {
+            }
+
+            public Address InputAddress => Address.Parse(SatelliteAddress);
+            public bool Disabled => false;
+        }
+    }
+}

src/NServiceBus.SqlServer.UnitTests/AdaptiveExecutorSimulator/Simulator.cs

@@ -14,6 +14,7 @@ class Simulator : ISimulator
         List<Tuple<long, int, string>> result = new List<Tuple<long, int, string>>();
         List<Action> currentQueue = new List<Action>();
         int threadCounter;
+        // ReSharper disable once NotAccessedField.Local
         int processedMessages;
         private readonly List<Message> queue = new List<Message>();
 

src/NServiceBus.SqlServer/TableBasedQueue.cs

@@ -62,13 +62,14 @@ static object[] ExtractTransportMessageData(TransportMessage message, SendOption
 
             data[IdColumn] = Guid.Parse(message.Id);
             data[CorrelationIdColumn] = GetValue(message.CorrelationId);
+            string replyToAddress;
             if (sendOptions.ReplyToAddress != null)
             {
                 data[ReplyToAddressColumn] = sendOptions.ReplyToAddress.ToString();
             }
-            else if (message.ReplyToAddress != null)
+            else if (message.Headers.TryGetValue(Headers.ReplyToAddress, out replyToAddress))
             {
-                data[ReplyToAddressColumn] = message.ReplyToAddress.ToString();
+                data[ReplyToAddressColumn] = replyToAddress;
             }
             else
             {