Merge branch 'hotfix-5.2.2'

src/NServiceBus.AcceptanceTests/NServiceBus.AcceptanceTests.csproj

@@ -87,6 +87,7 @@
     <Compile Include="PubSub\When_publishing_from_sendonly.cs" />
     <Compile Include="PubSub\When_publishing_an_event_implementing_two_unrelated_interfaces.cs" />
     <Compile Include="Sagas\When_sagas_cant_be_found.cs" />
+    <Compile Include="Sagas\When_using_ReplyToOriginator.cs" />
     <Compile Include="ScaleOut\When_individualization_is_enabled_for_msmq.cs" />
     <Compile Include="ScaleOut\When_no_discriminator_is_available.cs" />
     <Compile Include="ScaleOut\When_individualization_is_enabled.cs" />

src/NServiceBus.AcceptanceTests/Sagas/When_using_ReplyToOriginator.cs

@@ -0,0 +1,97 @@
+namespace NServiceBus.AcceptanceTests.Sagas
+{
+    using System;
+    using EndpointTemplates;
+    using AcceptanceTesting;
+    using NUnit.Framework;
+    using Saga;
+
+    public class When_using_ReplyToOriginator : NServiceBusAcceptanceTest
+    {
+        [Test]
+        public void Should_set_Reply_as_messageintent()
+        {
+            var context = new Context();
+
+            Scenario.Define(context)
+                .WithEndpoint<Endpoint>(b => b.Given(bus => bus.SendLocal(new InitiateRequestingSaga())))
+                .Done(c => c.Done)
+                .Run();
+
+            Assert.AreEqual(MessageIntentEnum.Reply, context.Intent);
+        }
+
+        public class Context : ScenarioContext
+        {
+            public MessageIntentEnum Intent { get; set; }
+            public bool Done { get; set; }
+        }
+
+        public class Endpoint : EndpointConfigurationBuilder
+        {
+
+            public Endpoint()
+            {
+                EndpointSetup<DefaultServer>();
+            }
+
+            public class RequestingSaga : Saga<RequestingSaga.RequestingSagaData>,
+                IAmStartedByMessages<InitiateRequestingSaga>,
+                IHandleMessages<AnotherRequest>
+            {
+                public Context Context { get; set; }
+
+                public void Handle(InitiateRequestingSaga message)
+                {
+                    Data.CorrIdForResponse = Guid.NewGuid(); //wont be needed in the future
+
+                    Bus.SendLocal(new AnotherRequest
+                    {
+                        SomeCorrelationId = Data.CorrIdForResponse //wont be needed in the future
+                    });
+                }
+
+                public void Handle(AnotherRequest message)
+                {
+                    ReplyToOriginator(new MyReplyToOriginator());
+                    MarkAsComplete();
+                }
+
+                protected override void ConfigureHowToFindSaga(SagaPropertyMapper<RequestingSagaData> mapper)
+                {
+                    //if this line is un-commented the timeout and secondary handler tests will start to fail
+                    // for more info and discussion see TBD
+                    mapper.ConfigureMapping<AnotherRequest>(m => m.SomeCorrelationId).ToSaga(s => s.CorrIdForResponse);
+                }
+                public class RequestingSagaData : ContainSagaData
+                {
+                    public virtual Guid CorrIdForResponse { get; set; } //wont be needed in the future
+                }
+            }
+
+            class MyReplyToOriginatorHandler : IHandleMessages<MyReplyToOriginator>
+            {
+                public Context Context { get; set; }
+                public IBus Bus { get; set; }
+
+                public void Handle(MyReplyToOriginator message)
+                {
+                    Context.Intent = (MessageIntentEnum)Enum.Parse(typeof(MessageIntentEnum), Bus.CurrentMessageContext.Headers[Headers.MessageIntent]);
+                    Context.Done = true;
+                }
+            }
+        }
+
+        public class InitiateRequestingSaga : ICommand { }
+
+        public class AnotherRequest : ICommand
+        {
+            public Guid SomeCorrelationId { get; set; }
+        }
+
+        public class MyReplyToOriginator : IMessage
+        {
+            public Guid SomeCorrelationId { get; set; }
+        }
+    }
+}

src/NServiceBus.Core/NServiceBus.Core.csproj

@@ -239,6 +239,7 @@
     <Compile Include="Settings\DurableMessagesConfig.cs" />
     <Compile Include="Timeout\ConfigurationBuilderExtensions.cs" />
     <Compile Include="Transports\ConfigurePurging_Obsolete.cs" />
+    <Compile Include="Unicast\Behaviors\FixSendIntentBehavior.cs" />
     <Compile Include="Unicast\Behaviors\ForwardReceivedMessages.cs" />
     <Compile Include="Unicast\Behaviors\InvokeSagaNotFoundBehavior.cs" />
     <Compile Include="Unicast\BusAsyncResultEventArgs.cs" />

src/NServiceBus.Core/Pipeline/PipelineBuilder.cs

@@ -68,6 +68,7 @@ void RegisterOutgoingCoreBehaviors()
             coordinator.Register(WellKnownStep.MutateOutgoingMessages, typeof(MutateOutgoingMessageBehavior), "Executes IMutateOutgoingMessages");
             coordinator.Register("PopulateAutoCorrelationHeadersForReplies", typeof(PopulateAutoCorrelationHeadersForRepliesBehavior), "Copies existing saga headers from incoming message to outgoing message to facilitate the auto correlation in the saga, when replying to a message that was sent by a saga.");
             coordinator.Register(WellKnownStep.CreatePhysicalMessage, typeof(CreatePhysicalMessageBehavior), "Converts a logical message into a physical message");
+            coordinator.Register("FixSendIntent", typeof(FixSendIntentBehavior), "Fixes an issue where ReplyToOriginator isn't able to set the Reply message intent, breaking callbacks");
             coordinator.Register(WellKnownStep.SerializeMessage, typeof(SerializeMessagesBehavior), "Serializes the message to be sent out on the wire");
             coordinator.Register(WellKnownStep.MutateOutgoingTransportMessage, typeof(MutateOutgoingPhysicalMessageBehavior), "Executes IMutateOutgoingTransportMessages");
             if (LogManager.GetLogger("LogOutgoingMessage").IsDebugEnabled)

src/NServiceBus.Core/Saga/Saga.cs

@@ -155,6 +155,7 @@ protected virtual void ReplyToOriginator(object message)
             {
                 throw new Exception("Entity.Originator cannot be null. Perhaps the sender is a SendOnly endpoint.");
             }
+            Bus.SetMessageHeader(message, "$.temporary.ReplyToOriginator", "ReplyToOriginator"); // issue 2748
             Bus.Send(Entity.Originator, Entity.OriginalMessageId, message);
         }
 

src/NServiceBus.Core/Unicast/Behaviors/FixSendIntentBehavior.cs

@@ -0,0 +1,21 @@
+namespace NServiceBus
+{
+    using System;
+    using NServiceBus.Pipeline;
+    using NServiceBus.Pipeline.Contexts;
+
+    class FixSendIntentBehavior : IBehavior<OutgoingContext>
+    {
+        public void Invoke(OutgoingContext context, Action next)
+        {
+            if (context.OutgoingLogicalMessage.Headers.ContainsKey("$.temporary.ReplyToOriginator"))
+            {
+                context.OutgoingMessage.MessageIntent = MessageIntentEnum.Reply;
+                context.OutgoingMessage.Headers.Remove("$.temporary.ReplyToOriginator");
+                context.OutgoingLogicalMessage.Headers.Remove("$.temporary.ReplyToOriginator");
+            }
+
+            next();
+        }
+    }
+}