Merge branch 'hotfix/2.2.1' into develop

Tapeti.Flow/Default/FlowContinuationMiddleware.cs

@@ -52,16 +52,19 @@ namespace Tapeti.Flow.Default
         }
 
 
-        public async Task Cleanup(IMessageContext context, ConsumeResult consumeResult, Func<Task> next)
+        public async Task Cleanup(IControllerMessageContext context, ConsumeResult consumeResult, Func<Task> next)
         {
             await next();
 
             if (!context.Get(ContextItems.FlowContext, out FlowContext flowContext))
                 return;
 
+            if (flowContext.ContinuationMetadata.MethodName != MethodSerializer.Serialize(context.Binding.Method))
+                // Do not call when the controller method was filtered, if the same message has two methods
+                return;
+
             if (flowContext?.FlowStateLock != null)
             {
-                // TODO do not call when the controller method was filtered, if the same message has two methods
                 if (!flowContext.IsStoredOrDeleted())
                     // The exception strategy can set the consume result to Success. Instead, check if the yield point
                     // was handled. The flow provider ensures we only end up here in case of an exception.

Tapeti/Config/IControllerCleanupMiddleware.cs

@@ -14,6 +14,6 @@ namespace Tapeti.Config
         /// <param name="context"></param>
         /// <param name="consumeResult"></param>
         /// <param name="next">Always call to allow the next in the chain to clean up</param>
-        Task Cleanup(IMessageContext context, ConsumeResult consumeResult, Func<Task> next);
+        Task Cleanup(IControllerMessageContext context, ConsumeResult consumeResult, Func<Task> next);
     }
 }

Tapeti/Config/IControllerMessageContext.cs

@@ -7,7 +7,7 @@
     public interface IControllerMessageContext : IMessageContext
     {
         /// <summary>
-        /// An instance of the controller referenced by the binding.
+        /// An instance of the controller referenced by the binding. Note: is null during Cleanup.
         /// </summary>
         object Controller { get; }
 

Tapeti/Default/ControllerMethodBinding.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
@@ -182,10 +182,16 @@ namespace Tapeti.Default
         /// <inheritdoc />
         public async Task Cleanup(IMessageContext context, ConsumeResult consumeResult)
         {
-            await MiddlewareHelper.GoAsync(
+            using (var controllerContext = new ControllerMessageContext(context)
-                bindingInfo.CleanupMiddleware,
+            {
-                async (handler, next) => await handler.Cleanup(context, consumeResult, next),
+                Controller = null
-                () => Task.CompletedTask);
+            })
+            {
+                await MiddlewareHelper.GoAsync(
+                    bindingInfo.CleanupMiddleware,
+                    async (handler, next) => await handler.Cleanup(controllerContext, consumeResult, next),
+                    () => Task.CompletedTask);
+            }
         }
 
 

Tapeti/TapetiConfigControllers.cs

@@ -47,15 +47,8 @@ namespace Tapeti
                 .Where(m => m.MemberType == MemberTypes.Method && m.DeclaringType != typeof(object) && (m as MethodInfo)?.IsSpecialName == false)
                 .Select(m => (MethodInfo)m))
             {
-                var methodQueueInfo = GetQueueInfo(method) ?? controllerQueueInfo;
-                if (methodQueueInfo == null || !methodQueueInfo.IsValid)
-                    throw new TopologyConfigurationException(
-                        $"Method {method.Name} or controller {controller.Name} requires a queue attribute");
-
-
                 var methodIsObsolete = controllerIsObsolete || method.GetCustomAttribute<ObsoleteAttribute>() != null;
 
-
                 var context = new ControllerBindingContext(method.GetParameters(), method.ReturnParameter)
                 {
                     Controller = controller,
@@ -85,6 +78,10 @@ namespace Tapeti
                     throw new TopologyConfigurationException($"Method {method.Name} in controller {method.DeclaringType?.Name} has unknown parameters: {parameterNames}");
                 }
 
+                var methodQueueInfo = GetQueueInfo(method) ?? controllerQueueInfo;
+                if (methodQueueInfo == null || !methodQueueInfo.IsValid)
+                    throw new TopologyConfigurationException(
+                        $"Method {method.Name} or controller {controller.Name} requires a queue attribute");
 
                 builder.RegisterBinding(new ControllerMethodBinding(builderAccess.DependencyResolver, new ControllerMethodBinding.BindingInfo
                 {