1
0
mirror of synced 2024-11-13 06:19:16 +00:00

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

Fixed Continuation methods binding to dynamic queues
This commit is contained in:
Mark van Renswoude 2017-02-15 22:05:01 +01:00
parent 4a5b9673bd
commit 20ac467006
19 changed files with 201 additions and 36 deletions

View File

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

View File

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

View File

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

View File

@ -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);

View File

@ -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;

View File

@ -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;
}
}
}

View File

@ -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>();

View File

@ -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);

View File

@ -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" />

View File

@ -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);
}

View File

@ -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; }

View File

@ -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; }

View File

@ -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();
}
}
}
}

View File

@ -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);
}
}

View File

@ -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();
}
}
}

View File

@ -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" />

View File

@ -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;

View File

@ -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.

View File

@ -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();
}