Fixed #6: Provide a way to start a flow outside of a message handler

Fixed Continuation methods binding to dynamic queues

Tapeti.Annotations/RequestAttribute.cs

@@ -2,6 +2,7 @@
 
 namespace Tapeti.Flow.Annotations
 {
+    [AttributeUsage(AttributeTargets.Class)]
     public class RequestAttribute : Attribute
     {
         public Type Response { get; set; }

Tapeti.Flow/Annotations/ContinuationAttribute.cs

@@ -2,6 +2,7 @@
 
 namespace Tapeti.Flow.Annotations
 {
+    [AttributeUsage(AttributeTargets.Method)]
     public class ContinuationAttribute : Attribute
     {
     }

Tapeti.Flow/Annotations/StartAttribute.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace Tapeti.Flow.Annotations
+{
+    [AttributeUsage(AttributeTargets.Method)]
+    public class StartAttribute : Attribute
+    {
+    }
+}

Tapeti.Flow/Default/FlowBindingMiddleware.cs

@@ -7,11 +7,16 @@ using Tapeti.Helpers;
 
 namespace Tapeti.Flow.Default
 {
-    // TODO figure out a way to prevent binding on Continuation methods (which are always the target of a direct response)
     internal class FlowBindingMiddleware : IBindingMiddleware
     {
         public void Handle(IBindingContext context, Action next)
         {
+            if (context.Method.GetCustomAttribute<StartAttribute>() != null)
+                return;
+
+            if (context.Method.GetCustomAttribute<ContinuationAttribute>() != null)
+                context.QueueBindingMode = QueueBindingMode.DirectToQueue;
+
             RegisterYieldPointResult(context);
             RegisterContinuationFilter(context);
 

Tapeti.Flow/Default/FlowProvider.cs

@@ -119,7 +119,10 @@ namespace Tapeti.Flow.Default
 
             var continuationAttribute = binding.Method.GetCustomAttribute<ContinuationAttribute>();
             if (continuationAttribute == null)
-                throw new ArgumentException($"responseHandler must be marked with the Continuation attribute", nameof(responseHandler));
+                throw new ArgumentException("responseHandler must be marked with the Continuation attribute", nameof(responseHandler));
+
+            if (binding.QueueName == null)
+                throw new ArgumentException("responseHandler must bind to a valid queue", nameof(responseHandler));
 
             return new ResponseHandlerInfo
             {
@@ -131,7 +134,7 @@ namespace Tapeti.Flow.Default
 
         private static ReplyMetadata GetReply(IMessageContext context)
         {
-            var requestAttribute = context.Message.GetType().GetCustomAttribute<RequestAttribute>();
+            var requestAttribute = context.Message?.GetType().GetCustomAttribute<RequestAttribute>();
             if (requestAttribute?.Response == null)
                 return null;
 

Tapeti.Flow/Default/FlowStarter.cs

@@ -0,0 +1,61 @@
+using System;
+using System.Linq.Expressions;
+using System.Reflection;
+using System.Threading.Tasks;
+using Tapeti.Config;
+using Tapeti.Default;
+
+namespace Tapeti.Flow.Default
+{
+    public class FlowStarter : IFlowStarter
+    {
+        private readonly IConfig config;
+
+
+        public FlowStarter(IConfig config)
+        {
+            this.config = config;
+        }
+
+
+        public Task Start<TController>(Expression<Func<TController, Func<IYieldPoint>>> methodSelector) where TController : class
+        {
+            return CallControllerMethod<TController>(GetExpressionMethod(methodSelector), value => Task.FromResult((IYieldPoint)value));
+        }
+
+
+        public Task Start<TController>(Expression<Func<TController, Func<Task<IYieldPoint>>>> methodSelector) where TController : class
+        {
+            return CallControllerMethod<TController>(GetExpressionMethod(methodSelector), value => (Task<IYieldPoint>)value);
+        }
+
+
+        private async Task CallControllerMethod<TController>(MethodInfo method, Func<object, Task<IYieldPoint>> getYieldPointResult) where TController : class
+        {
+            var controller = config.DependencyResolver.Resolve<TController>();
+            var yieldPoint = await getYieldPointResult(method.Invoke(controller, new object[] {}));
+
+            var context = new MessageContext
+            {
+                DependencyResolver = config.DependencyResolver,
+                Controller = controller
+            };
+
+            var flowHandler = config.DependencyResolver.Resolve<IFlowHandler>();
+            await flowHandler.Execute(context, yieldPoint);
+        }
+
+
+        private static MethodInfo GetExpressionMethod<TController, TResult>(Expression<Func<TController, Func<TResult>>> methodSelector)
+        {
+            var callExpression = (methodSelector.Body as UnaryExpression)?.Operand as MethodCallExpression;
+            var targetMethodExpression = callExpression?.Object as ConstantExpression;
+
+            var method = targetMethodExpression?.Value as MethodInfo;
+            if (method == null)
+                throw new ArgumentException("Unable to determine the starting method", nameof(methodSelector));
+
+            return method;
+        }
+    }
+}

Tapeti.Flow/FlowMiddleware.cs

@@ -9,6 +9,7 @@ namespace Tapeti.Flow
         public void RegisterDefaults(IDependencyContainer container)
         {
             container.RegisterDefault<IFlowProvider, FlowProvider>();
+            container.RegisterDefault<IFlowStarter, FlowStarter>();
             container.RegisterDefault<IFlowHandler, FlowProvider>();
             container.RegisterDefault<IFlowRepository, NonPersistentFlowRepository>();
             container.RegisterDefault<IFlowStore, FlowStore>();

Tapeti.Flow/IFlowProvider.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Linq.Expressions;
 using System.Threading.Tasks;
 using Tapeti.Config;
 
@@ -19,6 +20,18 @@ namespace Tapeti.Flow
         IYieldPoint End();
     }
 
+    /// <summary>
+    /// Allows starting a flow outside of a message handler.
+    /// </summary>
+    public interface IFlowStarter
+    {
+        Task Start<TController>(Expression<Func<TController, Func<IYieldPoint>>> methodSelector) where TController : class;
+        Task Start<TController>(Expression<Func<TController, Func<Task<IYieldPoint>>>> methodSelector) where TController : class;
+    }
+
+    /// <summary>
+    /// Internal interface. Do not call directly.
+    /// </summary>
     public interface IFlowHandler
     {
         Task Execute(IMessageContext context, IYieldPoint yieldPoint);

Tapeti.Flow/Tapeti.Flow.csproj

@@ -52,11 +52,13 @@
   </ItemGroup>
   <ItemGroup>
     <Compile Include="Annotations\ContinuationAttribute.cs" />
+    <Compile Include="Annotations\StartAttribute.cs" />
     <Compile Include="ContextItems.cs" />
     <Compile Include="Default\FlowMessageFilterMiddleware.cs" />
     <Compile Include="Default\FlowBindingMiddleware.cs" />
     <Compile Include="Default\FlowContext.cs" />
     <Compile Include="Default\FlowMessageMiddleware.cs" />
+    <Compile Include="Default\FlowStarter.cs" />
     <Compile Include="Default\FlowState.cs" />
     <Compile Include="Default\IInternalYieldPoint.cs" />
     <Compile Include="Default\NonPersistentFlowRepository.cs" />

Tapeti/Config/IBindingContext.cs

@@ -9,6 +9,20 @@ namespace Tapeti.Config
     public delegate Task ResultHandler(IMessageContext context, object value);
 
 
+    public enum QueueBindingMode
+    {
+        /// <summary>
+        /// Allow binding of the routing key from the message's source exchange to the queue
+        /// </summary>
+        RoutingKey,
+
+        /// <summary>
+        /// Do not bind, rely on the direct-to-queue exchange
+        /// </summary>
+        DirectToQueue
+    }
+
+
     public interface IBindingContext
     {
         Type MessageClass { get; set; }
@@ -17,6 +31,8 @@ namespace Tapeti.Config
         IReadOnlyList<IBindingParameter> Parameters { get; }
         IBindingResult Result { get; }
 
+        QueueBindingMode QueueBindingMode { get; set; }
+
         void Use(IMessageFilterMiddleware filterMiddleware);
         void Use(IMessageMiddleware middleware);
     }

Tapeti/Config/IConfig.cs

@@ -37,6 +37,7 @@ namespace Tapeti.Config
         MethodInfo Method { get; }
         Type MessageClass { get; }
         string QueueName { get; }
+        QueueBindingMode QueueBindingMode { get; set; }
 
         IReadOnlyList<IMessageFilterMiddleware> MessageFilterMiddleware { get; }
         IReadOnlyList<IMessageMiddleware> MessageMiddleware { get; }

Tapeti/Config/IMessageContext.cs

@@ -15,9 +15,9 @@ namespace Tapeti.Config
 
         IDictionary<string, object> Items { get; }
 
-        /// <summary>
+        /// <remarks>
         /// Controller will be null when passed to a IMessageFilterMiddleware
-        /// </summary>
+        /// </remarks>
         object Controller { get; }
 
         IBinding Binding { get; }

Tapeti/Connection/TapetiConsumer.cs

@@ -4,6 +4,7 @@ using System.Linq;
 using System.Runtime.ExceptionServices;
 using RabbitMQ.Client;
 using Tapeti.Config;
+using Tapeti.Default;
 using Tapeti.Helpers;
 
 namespace Tapeti.Connection
@@ -118,28 +119,5 @@ namespace Tapeti.Connection
                 exception = aggregateException.InnerExceptions[0];
             }
         }
-
-
-        protected class MessageContext : IMessageContext
-        {
-            public IDependencyResolver DependencyResolver { get; set; }
-
-            public object Controller { get; set; }
-            public IBinding Binding { get; set; }
-
-            public string Queue { get; set; }
-            public string RoutingKey { get; set; }
-            public object Message { get; set;  }
-            public IBasicProperties Properties { get; set;  }
-
-            public IDictionary<string, object> Items { get; } = new Dictionary<string, object>();
-
-
-            public void Dispose()
-            {
-                foreach (var value in Items.Values)
-                    (value as IDisposable)?.Dispose();
-            }
-        }
     }
 }

Tapeti/Connection/TapetiWorker.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Exceptions;
@@ -70,10 +71,14 @@ namespace Tapeti.Connection
 
                     foreach (var binding in queue.Bindings)
                     {
-                        var routingKey = routingKeyStrategy.GetRoutingKey(binding.MessageClass);
-                        var exchange = exchangeStrategy.GetExchange(binding.MessageClass);
+                        if (binding.QueueBindingMode == QueueBindingMode.RoutingKey)
+                        {
+                            var routingKey = routingKeyStrategy.GetRoutingKey(binding.MessageClass);
+                            var exchange = exchangeStrategy.GetExchange(binding.MessageClass);
+
+                            channel.QueueBind(dynamicQueue.QueueName, exchange, routingKey);
+                        }
 
-                        channel.QueueBind(dynamicQueue.QueueName, exchange, routingKey);
                         (binding as IDynamicQueueBinding)?.SetQueueName(dynamicQueue.QueueName);
                     }
                 }

Tapeti/Default/MessageContext.cs

@@ -0,0 +1,29 @@
+using System;
+using System.Collections.Generic;
+using RabbitMQ.Client;
+using Tapeti.Config;
+
+namespace Tapeti.Default
+{
+    public class MessageContext : IMessageContext
+    {
+        public IDependencyResolver DependencyResolver { get; set; }
+
+        public object Controller { get; set; }
+        public IBinding Binding { get; set; }
+
+        public string Queue { get; set; }
+        public string RoutingKey { get; set; }
+        public object Message { get; set; }
+        public IBasicProperties Properties { get; set; }
+
+        public IDictionary<string, object> Items { get; } = new Dictionary<string, object>();
+
+
+        public void Dispose()
+        {
+            foreach (var value in Items.Values)
+                (value as IDisposable)?.Dispose();
+        }
+    }
+}

Tapeti/Tapeti.csproj

@@ -63,6 +63,7 @@
     <Compile Include="Default\ConsoleLogger.cs" />
     <Compile Include="Default\DevNullLogger.cs" />
     <Compile Include="Default\JsonMessageSerializer.cs" />
+    <Compile Include="Default\MessageContext.cs" />
     <Compile Include="Default\PublishResultBinding.cs" />
     <Compile Include="Default\NamespaceMatchExchangeStrategy.cs" />
     <Compile Include="Default\RequeueExceptionStrategy.cs" />

Tapeti/TapetiConfig.cs

@@ -145,18 +145,22 @@ namespace Tapeti
                 .Where(m => m.MemberType == MemberTypes.Method && m.DeclaringType != typeof(object))
                 .Select(m => (MethodInfo)m))
             {
-                var methodQueueInfo = GetQueueInfo(method) ?? controllerQueueInfo;
-                if (!methodQueueInfo.IsValid)
-                    throw new TopologyConfigurationException($"Method {method.Name} or controller {controller.Name} requires a queue attribute");
-
                 var context = new BindingContext(method);
                 var messageHandler = GetMessageHandler(context, method);
+                if (messageHandler == null)
+                    continue;
+
+                var methodQueueInfo = GetQueueInfo(method) ?? controllerQueueInfo;
+                if (!methodQueueInfo.IsValid)
+                    throw new TopologyConfigurationException(
+                        $"Method {method.Name} or controller {controller.Name} requires a queue attribute");
 
                 var handlerInfo = new Binding
                 {
                     Controller = controller,
                     Method = method,
                     QueueInfo = methodQueueInfo,
+                    QueueBindingMode = context.QueueBindingMode,
                     MessageClass = context.MessageClass,
                     MessageHandler = messageHandler,
                     MessageMiddleware = context.MessageMiddleware,
@@ -190,7 +194,17 @@ namespace Tapeti
 
         protected MessageHandlerFunc GetMessageHandler(IBindingContext context, MethodInfo method)
         {
-            MiddlewareHelper.Go(bindingMiddleware, (handler, next) => handler.Handle(context, next), () => {});
+            var allowBinding= false;
+
+            MiddlewareHelper.Go(bindingMiddleware, 
+                (handler, next) => handler.Handle(context, next),
+                () =>
+                {
+                    allowBinding = true;
+                });
+
+            if (!allowBinding)
+                return null;
 
             if (context.MessageClass == null)
                 throw new TopologyConfigurationException($"Method {method.Name} in controller {method.DeclaringType?.Name} does not resolve to a message class");
@@ -382,6 +396,7 @@ namespace Tapeti
             public MethodInfo Method { get; set; }
             public Type MessageClass { get; set; }
             public string QueueName { get; set; }
+            public QueueBindingMode QueueBindingMode { get; set; }
 
             public IReadOnlyList<IMessageMiddleware> MessageMiddleware { get; set;  }
             public IReadOnlyList<IMessageFilterMiddleware> MessageFilterMiddleware { get; set; }
@@ -443,6 +458,8 @@ namespace Tapeti
             public IReadOnlyList<IBindingParameter> Parameters { get; }
             public IBindingResult Result { get; }
 
+            public QueueBindingMode QueueBindingMode { get; set; }
+
             public IReadOnlyList<IMessageMiddleware> MessageMiddleware => messageMiddleware;
             public IReadOnlyList<IMessageFilterMiddleware> MessageFilterMiddleware => messageFilterMiddleware;
 

Test/MarcoController.cs

@@ -28,6 +28,26 @@ namespace Test
         }
 
 
+        [Start]
+        public async Task<IYieldPoint> StartFlow()
+        {
+            Console.WriteLine("Starting stand-alone flow");
+            await Task.Delay(1000);
+
+            return flowProvider.YieldWithRequestSync<PoloConfirmationRequestMessage, PoloConfirmationResponseMessage>
+                (new PoloConfirmationRequestMessage(),
+                HandlePoloConfirmationResponse);
+        }
+
+
+        [Continuation]
+        public IYieldPoint HandlePoloConfirmationResponse(PoloConfirmationResponseMessage msg)
+        {
+            Console.WriteLine("Ending stand-alone flow");
+            return flowProvider.End();
+        }
+
+
         /**
          * The Visualizer could've been injected through the constructor, which is
          * the recommended way. Just testing the injection middleware here.

Test/Program.cs

@@ -43,6 +43,8 @@ namespace Test
 
                 Console.WriteLine("Done!");
 
+                container.GetInstance<IFlowStarter>().Start<MarcoController>(c => c.StartFlow);
+
                 var emitter = container.GetInstance<MarcoEmitter>();
                 emitter.Run().Wait();
             }