Fixed MassTransit#4346 - TimeoutExpired for Request in State Machine …

…was using incorrect correlation syntax (also made annoying RMQ test Flaky)
RJSonnenberg · May 22, 2023 · 8bbb917 · 8bbb917
1 parent e078cd5
commit 8bbb917
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 7 deletions.
diff --git a/src/MassTransit/SagaStateMachine/MassTransitStateMachine.cs b/src/MassTransit/SagaStateMachine/MassTransitStateMachine.cs
@@ -1488,7 +1488,7 @@ protected internal void Request<TRequest, TResponse, TResponse2>(
             });
             Event(propertyExpression, x => x.TimeoutExpired, x =>
             {
-                x.CorrelateById(context => context.Message.RequestId);
+                x.CorrelateBy(requestIdExpression, context => context.Message.RequestId);
                 settings.TimeoutExpired?.Invoke(x);
             });
 
@@ -1664,7 +1664,7 @@ protected internal void Request<TRequest, TResponse, TResponse2, TResponse3>(
             });
             Event(propertyExpression, x => x.TimeoutExpired, x =>
             {
-                x.CorrelateById(context => context.Message.RequestId);
+                x.CorrelateBy(requestIdExpression, context => context.Message.RequestId);
                 settings.TimeoutExpired?.Invoke(x);
             });
 

diff --git a/tests/MassTransit.QuartzIntegration.Tests/Request_Specs.cs b/tests/MassTransit.QuartzIntegration.Tests/Request_Specs.cs
@@ -91,12 +91,12 @@ protected override void ConfigureInMemoryReceiveEndpoint(IInMemoryReceiveEndpoin
 
             protected virtual void ConfigureServiceQueueEndpoint(IReceiveEndpointConfigurator configurator)
             {
-                configurator.Handler<ValidateAddress>(async context =>
+                configurator.Handler<ValidateAddress>(context =>
                 {
                     Console.WriteLine("Address validated: {0}", context.Message.CorrelationId);
 
                     if (context.IsResponseAccepted<AddressValidated>(false))
-                        await context.RespondAsync<AddressValidated>(new { });
+                        return context.RespondAsync<AddressValidated>(new { });
 
                     throw new InvalidOperationException("Response type not accepted");
                 });
@@ -112,7 +112,7 @@ protected virtual void ConfigureServiceQueueEndpoint(IReceiveEndpointConfigurato
 
 
         class RequestSettingsImpl :
-            RequestSettings<TestState, ValidateAddress, AddressValidated>
+            RequestSettings<TestState, ValidateAddress, AddressValidated, AddressInvalidated>
         {
             public RequestSettingsImpl(Uri serviceAddress, Uri schedulingServiceAddress, TimeSpan timeout)
             {
@@ -128,6 +128,7 @@ public RequestSettingsImpl(Uri serviceAddress, Uri schedulingServiceAddress, Tim
             public TimeSpan Timeout { get; }
             public TimeSpan? TimeToLive { get; }
             public Action<IEventCorrelationConfigurator<TestState, AddressValidated>> Completed { get; set; }
+            public Action<IEventCorrelationConfigurator<TestState, AddressInvalidated>> Completed2 { get; set; }
             public Action<IEventCorrelationConfigurator<TestState, Fault<ValidateAddress>>> Faulted { get; set; }
             public Action<IEventCorrelationConfigurator<TestState, RequestTimeoutExpired<ValidateAddress>>> TimeoutExpired { get; set; }
         }
@@ -180,6 +181,15 @@ public interface AddressValidated :
         }
 
 
+        public interface AddressInvalidated :
+            CorrelatedBy<Guid>
+        {
+            string Address { get; }
+
+            string RequestAddress { get; }
+        }
+
+
         public interface ValidateName :
             CorrelatedBy<Guid>
         {
@@ -199,7 +209,7 @@ public interface NameValidated :
         class TestStateMachine :
             MassTransitStateMachine<TestState>
         {
-            public TestStateMachine(RequestSettings<TestState, ValidateAddress, AddressValidated> settings)
+            public TestStateMachine(RequestSettings<TestState, ValidateAddress, AddressValidated, AddressInvalidated> settings)
             {
                 Event(() => Register, x =>
                 {
@@ -261,7 +271,8 @@ public TestStateMachine(RequestSettings<TestState, ValidateAddress, AddressValid
                         .ThenAsync(async context => await Console.Out.WriteLineAsync("Request timed out"))
                         .TransitionTo(NameValidationTimeout));
             } // ReSharper disable UnassignedGetOnlyAutoProperty
-            public Request<TestState, ValidateAddress, AddressValidated> ValidateAddress { get; }
+
+            public Request<TestState, ValidateAddress, AddressValidated, AddressInvalidated> ValidateAddress { get; }
             public Request<TestState, ValidateName, NameValidated> ValidateName { get; }
 
             public Event<RegisterMember> Register { get; }

diff --git a/tests/MassTransit.RabbitMqTransport.Tests/TopologyCorrelationId_Specs.cs b/tests/MassTransit.RabbitMqTransport.Tests/TopologyCorrelationId_Specs.cs
@@ -6,6 +6,7 @@ namespace MassTransit.RabbitMqTransport.Tests
 
 
     [TestFixture]
+    [Category("Flaky")]
     public class When_using_a_name_property_for_correlation :
         RabbitMqTestFixture
     {
@@ -76,6 +77,7 @@ public class LegacyMessage
 
 
     [TestFixture]
+    [Category("Flaky")]
     public class When_using_a_base_event :
         RabbitMqTestFixture
     {