diff --git a/Tapeti.Annotations/DurableQueueAttribute.cs b/Tapeti.Annotations/DurableQueueAttribute.cs index 281d91f..8971044 100644 --- a/Tapeti.Annotations/DurableQueueAttribute.cs +++ b/Tapeti.Annotations/DurableQueueAttribute.cs @@ -14,9 +14,12 @@ namespace Tapeti.Annotations /// for deploy-time management of durable queues (shameless plug intended). /// [AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)] - [MeansImplicitUse] + [MeansImplicitUse(ImplicitUseTargetFlags.WithMembers)] public class DurableQueueAttribute : Attribute { + /// + /// Specifies the name of the durable queue (must already be declared). + /// public string Name { get; set; } diff --git a/Tapeti.Annotations/DynamicQueueAttribute.cs b/Tapeti.Annotations/DynamicQueueAttribute.cs index 3743edf..240b001 100644 --- a/Tapeti.Annotations/DynamicQueueAttribute.cs +++ b/Tapeti.Annotations/DynamicQueueAttribute.cs @@ -12,6 +12,11 @@ namespace Tapeti.Annotations [MeansImplicitUse] public class DynamicQueueAttribute : Attribute { + /// + /// An optional prefix. If specified, Tapeti will compose the queue name using the + /// prefix and a unique ID. If not specified, an empty queue name will be passed + /// to RabbitMQ thus letting it create a unique queue name. + /// public string Prefix { get; set; } diff --git a/Tapeti.Annotations/MessageControllerAttribute.cs b/Tapeti.Annotations/MessageControllerAttribute.cs index 150fefc..6a18416 100644 --- a/Tapeti.Annotations/MessageControllerAttribute.cs +++ b/Tapeti.Annotations/MessageControllerAttribute.cs @@ -9,7 +9,7 @@ namespace Tapeti.Annotations /// when using the RegisterAllControllers method. It is not required when manually registering a controller. /// [AttributeUsage(AttributeTargets.Class)] - [MeansImplicitUse] + [MeansImplicitUse(ImplicitUseTargetFlags.WithMembers)] public class MessageControllerAttribute : Attribute { } diff --git a/Tapeti.Annotations/MessageHandlerAttribute.cs b/Tapeti.Annotations/MessageHandlerAttribute.cs deleted file mode 100644 index d13724e..0000000 --- a/Tapeti.Annotations/MessageHandlerAttribute.cs +++ /dev/null @@ -1,17 +0,0 @@ -using System; -using JetBrains.Annotations; - -namespace Tapeti.Annotations -{ - /// - /// - /// This attribute does nothing in runtime and is not required. It is only used as - /// a hint to ReSharper, and maybe developers as well, to indicate the method is - /// indeed used. - /// - [AttributeUsage(AttributeTargets.Method)] - [MeansImplicitUse] - public class MessageHandlerAttribute : Attribute - { - } -} diff --git a/Tapeti.Annotations/RequestAttribute.cs b/Tapeti.Annotations/RequestAttribute.cs index 2f14097..f298c50 100644 --- a/Tapeti.Annotations/RequestAttribute.cs +++ b/Tapeti.Annotations/RequestAttribute.cs @@ -13,6 +13,9 @@ namespace Tapeti.Annotations [AttributeUsage(AttributeTargets.Class)] public class RequestAttribute : Attribute { + /// + /// The type of the message class which must be returned as the response. + /// public Type Response { get; set; } } } diff --git a/Tapeti.Annotations/Tapeti.Annotations.csproj b/Tapeti.Annotations/Tapeti.Annotations.csproj index 9f5c4f4..be5c9ef 100644 --- a/Tapeti.Annotations/Tapeti.Annotations.csproj +++ b/Tapeti.Annotations/Tapeti.Annotations.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti.DataAnnotations/Tapeti.DataAnnotations.csproj b/Tapeti.DataAnnotations/Tapeti.DataAnnotations.csproj index 3df5704..52e0d73 100644 --- a/Tapeti.DataAnnotations/Tapeti.DataAnnotations.csproj +++ b/Tapeti.DataAnnotations/Tapeti.DataAnnotations.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj b/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj index b80a796..eaa2e91 100644 --- a/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj +++ b/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti.Flow/Default/FlowProvider.cs b/Tapeti.Flow/Default/FlowProvider.cs index 289b246..5c6d9d9 100644 --- a/Tapeti.Flow/Default/FlowProvider.cs +++ b/Tapeti.Flow/Default/FlowProvider.cs @@ -105,9 +105,9 @@ namespace Tapeti.Flow.Default // TODO disallow if replyto is not specified? if (reply.ReplyTo != null) - await publisher.PublishDirect(message, reply.ReplyTo, properties, true); + await publisher.PublishDirect(message, reply.ReplyTo, properties, reply.Mandatory); else - await publisher.Publish(message, properties, true); + await publisher.Publish(message, properties, reply.Mandatory); await context.Delete(); } @@ -129,8 +129,8 @@ namespace Tapeti.Flow.Default throw new ArgumentException("responseHandler must be a registered message handler", nameof(responseHandler)); var requestAttribute = request.GetType().GetCustomAttribute(); - if (requestAttribute?.Response != null && requestAttribute.Response != binding.MessageClass) - throw new ArgumentException($"responseHandler must accept message of type {binding.MessageClass}", nameof(responseHandler)); + if (requestAttribute?.Response != null && !binding.Accept(requestAttribute.Response)) + throw new ArgumentException($"responseHandler must accept message of type {requestAttribute.Response}", nameof(responseHandler)); var continuationAttribute = binding.Method.GetCustomAttribute(); if (continuationAttribute == null) @@ -157,7 +157,8 @@ namespace Tapeti.Flow.Default { CorrelationId = context.Properties.CorrelationId, ReplyTo = context.Properties.ReplyTo, - ResponseTypeName = requestAttribute.Response.FullName + ResponseTypeName = requestAttribute.Response.FullName, + Mandatory = context.Properties.Persistent }; } diff --git a/Tapeti.Flow/Default/FlowState.cs b/Tapeti.Flow/Default/FlowState.cs index d600a8f..e32430c 100644 --- a/Tapeti.Flow/Default/FlowState.cs +++ b/Tapeti.Flow/Default/FlowState.cs @@ -57,6 +57,8 @@ namespace Tapeti.Flow.Default public string CorrelationId { get; set; } public string ResponseTypeName { get; set; } + public bool Mandatory { get; set; } + public ReplyMetadata Clone() { @@ -64,7 +66,8 @@ namespace Tapeti.Flow.Default { ReplyTo = ReplyTo, CorrelationId = CorrelationId, - ResponseTypeName = ResponseTypeName + ResponseTypeName = ResponseTypeName, + Mandatory = Mandatory }; } } diff --git a/Tapeti.Flow/Default/FlowStore.cs b/Tapeti.Flow/Default/FlowStore.cs index 2597007..b98923d 100644 --- a/Tapeti.Flow/Default/FlowStore.cs +++ b/Tapeti.Flow/Default/FlowStore.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading.Tasks; using Tapeti.Flow.FlowHelpers; @@ -16,6 +17,7 @@ namespace Tapeti.Flow.Default private readonly IFlowRepository repository; private volatile bool inUse; + private volatile bool loaded; public FlowStore(IFlowRepository repository) { @@ -40,17 +42,25 @@ namespace Tapeti.Flow.Default foreach (var continuation in flowStateRecord.Value.Continuations) continuationLookup.GetOrAdd(continuation.Key, flowStateRecord.Key); } + + loaded = true; } public Task FindFlowID(Guid continuationID) { + if (!loaded) + throw new InvalidOperationException("Flow store is not yet loaded."); + return Task.FromResult(continuationLookup.TryGetValue(continuationID, out var result) ? result : (Guid?)null); } public async Task LockFlowState(Guid flowID) { + if (!loaded) + throw new InvalidOperationException("Flow store should be loaded before storing flows."); + inUse = true; var flowStatelock = new FlowStateLock(this, flowID, await locks.GetLock(flowID)); diff --git a/Tapeti.Flow/Tapeti.Flow.csproj b/Tapeti.Flow/Tapeti.Flow.csproj index cc0d9d9..105aa14 100644 --- a/Tapeti.Flow/Tapeti.Flow.csproj +++ b/Tapeti.Flow/Tapeti.Flow.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti.Serilog/Tapeti.Serilog.csproj b/Tapeti.Serilog/Tapeti.Serilog.csproj index ec64a23..b33e71b 100644 --- a/Tapeti.Serilog/Tapeti.Serilog.csproj +++ b/Tapeti.Serilog/Tapeti.Serilog.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti.SimpleInjector/Tapeti.SimpleInjector.csproj b/Tapeti.SimpleInjector/Tapeti.SimpleInjector.csproj index 25ed29e..ed72a19 100644 --- a/Tapeti.SimpleInjector/Tapeti.SimpleInjector.csproj +++ b/Tapeti.SimpleInjector/Tapeti.SimpleInjector.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti.Tests/TransientFilterMiddleware.cs b/Tapeti.Tests/TransientFilterMiddleware.cs new file mode 100644 index 0000000..d311f03 --- /dev/null +++ b/Tapeti.Tests/TransientFilterMiddleware.cs @@ -0,0 +1,14 @@ +using System; +using System.Threading.Tasks; +using Tapeti.Config; + +namespace Tapeti.Tests +{ + public class TransientFilterMiddleware : IMessageFilterMiddleware + { + public Task Handle(IMessageContext context, Func next) + { + throw new NotImplementedException(); + } + } +} \ No newline at end of file diff --git a/Tapeti.Transient/ConfigExtentions.cs b/Tapeti.Transient/ConfigExtentions.cs new file mode 100644 index 0000000..7401578 --- /dev/null +++ b/Tapeti.Transient/ConfigExtentions.cs @@ -0,0 +1,13 @@ +using System; + +namespace Tapeti.Transient +{ + public static class ConfigExtensions + { + public static TapetiConfig WithTransient(this TapetiConfig config, TimeSpan defaultTimeout, string dynamicQueuePrefix = "transient") + { + config.Use(new TransientMiddleware(defaultTimeout, dynamicQueuePrefix)); + return config; + } + } +} \ No newline at end of file diff --git a/Tapeti.Transient/ITransientPublisher.cs b/Tapeti.Transient/ITransientPublisher.cs new file mode 100644 index 0000000..2765259 --- /dev/null +++ b/Tapeti.Transient/ITransientPublisher.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace Tapeti.Transient +{ + public interface ITransientPublisher + { + Task RequestResponse(TRequest request); + } +} \ No newline at end of file diff --git a/Tapeti.Transient/Tapeti.Transient.csproj b/Tapeti.Transient/Tapeti.Transient.csproj new file mode 100644 index 0000000..f3cca6f --- /dev/null +++ b/Tapeti.Transient/Tapeti.Transient.csproj @@ -0,0 +1,12 @@ + + + + netstandard2.0 + true + + + + + + + diff --git a/Tapeti.Transient/Tapeti.Transient.nuspec b/Tapeti.Transient/Tapeti.Transient.nuspec new file mode 100644 index 0000000..ad29af0 --- /dev/null +++ b/Tapeti.Transient/Tapeti.Transient.nuspec @@ -0,0 +1,23 @@ + + + + Tapeti.Transient + $version$ + Tapeti Transient + Menno van Lavieren, Mark van Renswoude + Mark van Renswoude + https://raw.githubusercontent.com/MvRens/Tapeti/master/UNLICENSE + https://github.com/MvRens/Tapeti + https://raw.githubusercontent.com/MvRens/Tapeti/master/resources/icons/Tapeti.Flow.png + false + Transient extension for Tapeti + + rabbitmq tapeti transient + + + + + + + + \ No newline at end of file diff --git a/Tapeti.Transient/TransientGenericBinding.cs b/Tapeti.Transient/TransientGenericBinding.cs new file mode 100644 index 0000000..f28643d --- /dev/null +++ b/Tapeti.Transient/TransientGenericBinding.cs @@ -0,0 +1,52 @@ +using System; +using System.Reflection; +using System.Threading.Tasks; +using Tapeti.Config; + +namespace Tapeti.Transient +{ + public class TransientGenericBinding : ICustomBinding + { + private readonly TransientRouter router; + + public TransientGenericBinding(TransientRouter router, string dynamicQueuePrefix) + { + this.router = router; + DynamicQueuePrefix = dynamicQueuePrefix; + Method = typeof(TransientRouter).GetMethod("GenericHandleResponse"); + } + + public Type Controller => typeof(TransientRouter); + + public MethodInfo Method { get; } + + public QueueBindingMode QueueBindingMode => QueueBindingMode.DirectToQueue; + + public string StaticQueueName => null; + + public string DynamicQueuePrefix { get; } + + public Type MessageClass => null; + + public bool Accept(Type messageClass) + { + return true; + } + + public bool Accept(IMessageContext context, object message) + { + return true; + } + + public Task Invoke(IMessageContext context, object message) + { + router.GenericHandleResponse(message, context); + return Task.CompletedTask; + } + + public void SetQueueName(string queueName) + { + router.TransientResponseQueueName = queueName; + } + } +} \ No newline at end of file diff --git a/Tapeti.Transient/TransientMiddleware.cs b/Tapeti.Transient/TransientMiddleware.cs new file mode 100644 index 0000000..5077fa5 --- /dev/null +++ b/Tapeti.Transient/TransientMiddleware.cs @@ -0,0 +1,34 @@ +using System; +using System.Collections.Generic; +using Tapeti.Config; + +namespace Tapeti.Transient +{ + public class TransientMiddleware : ITapetiExtension, ITapetiExtentionBinding + { + private string dynamicQueuePrefix; + private readonly TransientRouter router; + + public TransientMiddleware(TimeSpan defaultTimeout, string dynamicQueuePrefix) + { + this.dynamicQueuePrefix = dynamicQueuePrefix; + this.router = new TransientRouter(defaultTimeout); + } + + public void RegisterDefaults(IDependencyContainer container) + { + container.RegisterDefaultSingleton(router); + container.RegisterDefault(); + } + + public IEnumerable GetMiddleware(IDependencyResolver dependencyResolver) + { + return new object[0]; + } + + public IEnumerable GetBindings(IDependencyResolver dependencyResolver) + { + yield return new TransientGenericBinding(router, dynamicQueuePrefix); + } + } +} \ No newline at end of file diff --git a/Tapeti.Transient/TransientPublisher.cs b/Tapeti.Transient/TransientPublisher.cs new file mode 100644 index 0000000..62715e7 --- /dev/null +++ b/Tapeti.Transient/TransientPublisher.cs @@ -0,0 +1,24 @@ +using System; +using System.Collections.Generic; +using System.Text; +using System.Threading.Tasks; + +namespace Tapeti.Transient +{ + public class TransientPublisher : ITransientPublisher + { + private readonly TransientRouter router; + private readonly IPublisher publisher; + + public TransientPublisher(TransientRouter router, IPublisher publisher) + { + this.router = router; + this.publisher = publisher; + } + + public async Task RequestResponse(TRequest request) + { + return (TResponse)(await router.RequestResponse(publisher, request)); + } + } +} diff --git a/Tapeti.Transient/TransientRouter.cs b/Tapeti.Transient/TransientRouter.cs new file mode 100644 index 0000000..775576b --- /dev/null +++ b/Tapeti.Transient/TransientRouter.cs @@ -0,0 +1,72 @@ +using System; +using System.Collections.Concurrent; +using System.Threading.Tasks; +using System.Threading; +using RabbitMQ.Client.Framing; +using Tapeti.Config; + +namespace Tapeti.Transient +{ + public class TransientRouter + { + private readonly int defaultTimeoutMs; + + private readonly ConcurrentDictionary> map = new ConcurrentDictionary>(); + + + public string TransientResponseQueueName { get; set; } + + public TransientRouter(TimeSpan defaultTimeout) + { + defaultTimeoutMs = (int)defaultTimeout.TotalMilliseconds; + } + + public void GenericHandleResponse(object response, IMessageContext context) + { + if (context.Properties.CorrelationId == null) + return; + + if (!Guid.TryParse(context.Properties.CorrelationId, out var continuationID)) + return; + + if (map.TryRemove(continuationID, out var tcs)) + tcs.SetResult(response); + } + + public async Task RequestResponse(IPublisher publisher, object request) + { + var correlation = Guid.NewGuid(); + var tcs = map.GetOrAdd(correlation, c => new TaskCompletionSource()); + + try + { + var properties = new BasicProperties + { + CorrelationId = correlation.ToString(), + ReplyTo = TransientResponseQueueName, + Persistent = false + }; + + await ((IInternalPublisher)publisher).Publish(request, properties, false); + } + catch (Exception) + { + // Simple cleanup of the task and map dictionary. + if (map.TryRemove(correlation, out tcs)) + tcs.TrySetResult(null); + + throw; + } + + using (new Timer(TimeoutResponse, tcs, defaultTimeoutMs, -1)) + { + return await tcs.Task; + } + } + + private void TimeoutResponse(object tcs) + { + ((TaskCompletionSource)tcs).SetException(new TimeoutException("Transient RequestResponse timed out at (ms) " + defaultTimeoutMs)); + } + } +} \ No newline at end of file diff --git a/Tapeti.sln b/Tapeti.sln index 73f1486..114f0a9 100644 --- a/Tapeti.sln +++ b/Tapeti.sln @@ -19,7 +19,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Test", "Test\Test.csproj", EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Tests", "Tapeti.Tests\Tapeti.Tests.csproj", "{334F3715-63CF-4D13-B09A-38E2A616D4F5}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tapeti.Serilog", "Tapeti.Serilog\Tapeti.Serilog.csproj", "{43AA5DF3-49D5-4795-A290-D6511502B564}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Serilog", "Tapeti.Serilog\Tapeti.Serilog.csproj", "{43AA5DF3-49D5-4795-A290-D6511502B564}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tapeti.Transient", "Tapeti.Transient\Tapeti.Transient.csproj", "{A6355E63-19AB-47EA-91FA-49B5E9B41F88}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution @@ -63,6 +65,10 @@ Global {43AA5DF3-49D5-4795-A290-D6511502B564}.Debug|Any CPU.Build.0 = Debug|Any CPU {43AA5DF3-49D5-4795-A290-D6511502B564}.Release|Any CPU.ActiveCfg = Release|Any CPU {43AA5DF3-49D5-4795-A290-D6511502B564}.Release|Any CPU.Build.0 = Release|Any CPU + {A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Debug|Any CPU.Build.0 = Debug|Any CPU + {A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Release|Any CPU.ActiveCfg = Release|Any CPU + {A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/Tapeti/Config/IConfig.cs b/Tapeti/Config/IConfig.cs index 6f7449d..1cdaad7 100644 --- a/Tapeti/Config/IConfig.cs +++ b/Tapeti/Config/IConfig.cs @@ -46,6 +46,7 @@ namespace Tapeti.Config IReadOnlyList MessageFilterMiddleware { get; } IReadOnlyList MessageMiddleware { get; } + bool Accept(Type messageClass); bool Accept(IMessageContext context, object message); Task Invoke(IMessageContext context, object message); } diff --git a/Tapeti/Config/ICustomBinding.cs b/Tapeti/Config/ICustomBinding.cs new file mode 100644 index 0000000..8b39247 --- /dev/null +++ b/Tapeti/Config/ICustomBinding.cs @@ -0,0 +1,31 @@ +using System; +using System.Collections.Generic; +using System.Reflection; +using System.Text; +using System.Threading.Tasks; + +namespace Tapeti.Config +{ + public interface ICustomBinding + { + Type Controller { get; } + + MethodInfo Method { get; } + + QueueBindingMode QueueBindingMode { get; } + + string StaticQueueName { get; } + + string DynamicQueuePrefix { get; } + + Type MessageClass { get; } // Needed to get routing key information when QueueBindingMode = RoutingKey + + bool Accept(Type messageClass); + + bool Accept(IMessageContext context, object message); + + Task Invoke(IMessageContext context, object message); + + void SetQueueName(string queueName); + } +} diff --git a/Tapeti/Config/ITapetiExtentionBinding.cs b/Tapeti/Config/ITapetiExtentionBinding.cs new file mode 100644 index 0000000..5eee3a4 --- /dev/null +++ b/Tapeti/Config/ITapetiExtentionBinding.cs @@ -0,0 +1,10 @@ +using System.Collections.Generic; + +namespace Tapeti.Config +{ + public interface ITapetiExtentionBinding + { + IEnumerable GetBindings(IDependencyResolver dependencyResolver); + + } +} \ No newline at end of file diff --git a/Tapeti/Connection/TapetiWorker.cs b/Tapeti/Connection/TapetiWorker.cs index df0f3d7..2993535 100644 --- a/Tapeti/Connection/TapetiWorker.cs +++ b/Tapeti/Connection/TapetiWorker.cs @@ -40,7 +40,7 @@ namespace Tapeti.Connection private DateTime connectedDateTime; // These fields must be locked, since the callbacks for BasicAck/BasicReturn can run in a different thread - private readonly object confirmLock = new Object(); + private readonly object confirmLock = new object(); private readonly Dictionary confirmMessages = new Dictionary(); private readonly Dictionary returnRoutingKeys = new Dictionary(); @@ -113,6 +113,9 @@ namespace Tapeti.Connection { if (binding.QueueBindingMode == QueueBindingMode.RoutingKey) { + if (binding.MessageClass == null) + throw new NullReferenceException("Binding with QueueBindingMode = RoutingKey must have a MessageClass"); + var routingKey = routingKeyStrategy.GetRoutingKey(binding.MessageClass); var exchange = exchangeStrategy.GetExchange(binding.MessageClass); diff --git a/Tapeti/Default/PublishResultBinding.cs b/Tapeti/Default/PublishResultBinding.cs index 2ffe4e1..76542c3 100644 --- a/Tapeti/Default/PublishResultBinding.cs +++ b/Tapeti/Default/PublishResultBinding.cs @@ -69,7 +69,7 @@ namespace Tapeti.Default properties.CorrelationId = messageContext.Properties.CorrelationId; if (messageContext.Properties.IsReplyToPresent()) - return publisher.PublishDirect(message, messageContext.Properties.ReplyTo, properties, true); + return publisher.PublishDirect(message, messageContext.Properties.ReplyTo, properties, messageContext.Properties.Persistent); return publisher.Publish(message, properties, false); } diff --git a/Tapeti/Tapeti.csproj b/Tapeti/Tapeti.csproj index ecc266a..d4ecad3 100644 --- a/Tapeti/Tapeti.csproj +++ b/Tapeti/Tapeti.csproj @@ -2,6 +2,7 @@ netstandard2.0 + true diff --git a/Tapeti/TapetiConfig.cs b/Tapeti/TapetiConfig.cs index c785408..9291798 100644 --- a/Tapeti/TapetiConfig.cs +++ b/Tapeti/TapetiConfig.cs @@ -22,9 +22,11 @@ namespace Tapeti public class TapetiConfig { - private readonly Dictionary> staticRegistrations = new Dictionary>(); - private readonly Dictionary>> dynamicRegistrations = new Dictionary>>(); + private readonly Dictionary> staticRegistrations = new Dictionary>(); + private readonly Dictionary>> dynamicRegistrations = new Dictionary>>(); + private readonly List uniqueRegistrations = new List(); + private readonly List customBindings = new List(); private readonly List bindingMiddleware = new List(); private readonly List messageMiddleware = new List(); private readonly List cleanupMiddleware = new List(); @@ -47,6 +49,8 @@ namespace Tapeti public IConfig Build() { + RegisterCustomBindings(); + RegisterDefaults(); var queues = new List(); @@ -79,12 +83,12 @@ namespace Tapeti // foreach (var prefixGroup in dynamicRegistrations) { - var dynamicBindings = new List>(); + var dynamicBindings = new List>(); foreach (var bindings in prefixGroup.Value.Values) { while (dynamicBindings.Count < bindings.Count) - dynamicBindings.Add(new List()); + dynamicBindings.Add(new List()); for (var bindingIndex = 0; bindingIndex < bindings.Count; bindingIndex++) dynamicBindings[bindingIndex].Add(bindings[bindingIndex]); @@ -93,6 +97,9 @@ namespace Tapeti queues.AddRange(dynamicBindings.Select(bl => new Queue(new QueueInfo { Dynamic = true, Name = GetDynamicQueueName(prefixGroup.Key) }, bl))); } + queues.AddRange(uniqueRegistrations.Select(b => new Queue(new QueueInfo { Dynamic = true, Name = GetDynamicQueueName(b.QueueInfo.Name) }, new []{b}))); + + var config = new Config(queues) { DependencyResolver = dependencyResolver, @@ -144,6 +151,9 @@ namespace Tapeti var middlewareBundle = extension.GetMiddleware(dependencyResolver); + if (extension is ITapetiExtentionBinding extentionBindings) + customBindings.AddRange(extentionBindings.GetBindings(dependencyResolver)); + // ReSharper disable once InvertIf if (middlewareBundle != null) { @@ -212,7 +222,8 @@ namespace Tapeti { var controllerQueueInfo = GetQueueInfo(controller); - (dependencyResolver as IDependencyContainer)?.RegisterController(controller); + if (!controller.IsInterface) + (dependencyResolver as IDependencyContainer)?.RegisterController(controller); foreach (var method in controller.GetMembers(BindingFlags.Public | BindingFlags.Instance) .Where(m => m.MemberType == MemberTypes.Method && m.DeclaringType != typeof(object) && (m as MethodInfo)?.IsSpecialName == false) @@ -241,9 +252,9 @@ namespace Tapeti }; if (methodQueueInfo.Dynamic.GetValueOrDefault()) - AddDynamicRegistration(context, handlerInfo); + AddDynamicRegistration(handlerInfo); else - AddStaticRegistration(context, handlerInfo); + AddStaticRegistration(handlerInfo); } return this; @@ -264,6 +275,27 @@ namespace Tapeti return RegisterAllControllers(Assembly.GetEntryAssembly()); } + private void RegisterCustomBindings() + { + foreach (var customBinding in customBindings) + { + // TODO Do we need to configure additional middleware, or does this only get confused if there is no MessageClass + + var binding = new CustomBinding(customBinding); + if (binding.QueueInfo.Dynamic == false) + { + AddStaticRegistration(binding); + } + else if (binding.MessageClass != null) + { + AddDynamicRegistration(binding); + } + else + { + AddUniqueRegistration(binding); + } + } + } protected MessageHandlerFunc GetMessageHandler(IBindingContext context, MethodInfo method) { @@ -359,7 +391,7 @@ namespace Tapeti } - protected void AddStaticRegistration(IBindingContext context, Binding binding) + protected void AddStaticRegistration(IBindingQueueInfo binding) { if (staticRegistrations.ContainsKey(binding.QueueInfo.Name)) { @@ -372,29 +404,33 @@ namespace Tapeti existing.Add(binding); } else - staticRegistrations.Add(binding.QueueInfo.Name, new List { binding }); + staticRegistrations.Add(binding.QueueInfo.Name, new List { binding }); } - protected void AddDynamicRegistration(IBindingContext context, Binding binding) + protected void AddDynamicRegistration(IBindingQueueInfo binding) { var prefix = binding.QueueInfo.Name ?? ""; - if (!dynamicRegistrations.TryGetValue(prefix, out Dictionary> prefixRegistrations)) + if (!dynamicRegistrations.TryGetValue(prefix, out Dictionary> prefixRegistrations)) { - prefixRegistrations = new Dictionary>(); + prefixRegistrations = new Dictionary>(); dynamicRegistrations.Add(prefix, prefixRegistrations); } - if (!prefixRegistrations.TryGetValue(context.MessageClass, out List bindings)) + if (!prefixRegistrations.TryGetValue(binding.MessageClass, out List bindings)) { - bindings = new List(); - prefixRegistrations.Add(context.MessageClass, bindings); + bindings = new List(); + prefixRegistrations.Add(binding.MessageClass, bindings); } bindings.Add(binding); } + protected void AddUniqueRegistration(IBindingQueueInfo binding) + { + uniqueRegistrations.Add(binding); + } protected QueueInfo GetQueueInfo(MemberInfo member) { @@ -491,8 +527,12 @@ namespace Tapeti } } + protected interface IBindingQueueInfo : IBuildBinding + { + QueueInfo QueueInfo { get; } + } - protected class Binding : IBuildBinding + protected class Binding : IBindingQueueInfo { public Type Controller { get; set; } public MethodInfo Method { get; set; } @@ -523,6 +563,11 @@ namespace Tapeti } + public bool Accept(Type messageClass) + { + return MessageClass.IsAssignableFrom(messageClass); + } + public bool Accept(IMessageContext context, object message) { return message.GetType() == MessageClass; @@ -536,6 +581,69 @@ namespace Tapeti } + protected class CustomBinding : IBindingQueueInfo + { + private readonly ICustomBinding inner; + + public CustomBinding(ICustomBinding inner) + { + this.inner = inner; + + // Copy all variables to make them guaranteed readonly. + Controller = inner.Controller; + Method = inner.Method; + QueueBindingMode = inner.QueueBindingMode; + MessageClass = inner.MessageClass; + + QueueInfo = inner.StaticQueueName != null + ? new QueueInfo() + { + Dynamic = false, + Name = inner.StaticQueueName + } + : new QueueInfo() + { + Dynamic = true, + Name = inner.DynamicQueuePrefix + }; + + // Custom bindings cannot have other middleware messing with the binding. + MessageFilterMiddleware = new IMessageFilterMiddleware[0]; + MessageMiddleware = new IMessageMiddleware[0]; + } + + public Type Controller { get; } + public MethodInfo Method { get; } + public string QueueName { get; private set; } + public QueueBindingMode QueueBindingMode { get; set; } + public IReadOnlyList MessageFilterMiddleware { get; } + public IReadOnlyList MessageMiddleware { get; } + + public bool Accept(Type messageClass) + { + return inner.Accept(messageClass); + } + + public bool Accept(IMessageContext context, object message) + { + return inner.Accept(context, message); + } + + public Task Invoke(IMessageContext context, object message) + { + return inner.Invoke(context, message); + } + + public void SetQueueName(string queueName) + { + QueueName = queueName; + inner.SetQueueName(queueName); + } + + public Type MessageClass { get; } + public QueueInfo QueueInfo { get; } + } + internal interface IBindingParameterAccess { ValueFactory GetBinding(); diff --git a/Test/Program.cs b/Test/Program.cs index 62377aa..3969269 100644 --- a/Test/Program.cs +++ b/Test/Program.cs @@ -1,10 +1,13 @@ using System; +using System.Runtime.CompilerServices; using SimpleInjector; using Tapeti; using Tapeti.DataAnnotations; using Tapeti.Flow; using Tapeti.SimpleInjector; using System.Threading; +using Tapeti.Annotations; +using Tapeti.Transient; namespace Test { @@ -13,7 +16,7 @@ namespace Test private static void Main() { // TODO logging - try + //try { var container = new Container(); container.Register(); @@ -24,6 +27,7 @@ namespace Test //.WithFlowSqlRepository("Server=localhost;Database=TapetiTest;Integrated Security=true") .WithFlow() .WithDataAnnotations() + .WithTransient(TimeSpan.FromSeconds(30)) .RegisterAllControllers() //.DisablePublisherConfirms() -> you probably never want to do this if you're using Flow or want requeues when a publish fails .Build(); @@ -50,23 +54,32 @@ namespace Test Console.WriteLine("Done!"); - connection.GetPublisher().Publish(new FlowEndController.PingMessage()); + var response = container.GetInstance() + .RequestResponse( + new PoloConfirmationRequestMessage + { + StoredInState = new Guid("309088d8-9906-4ef3-bc64-56976538d3ab") + }).Result; + + Console.WriteLine(response.ShouldMatchState); + + //connection.GetPublisher().Publish(new FlowEndController.PingMessage()); //container.GetInstance().Start(c => c.StartFlow, true).Wait(); - container.GetInstance().Start(c => c.TestParallelRequest).Wait(); + //container.GetInstance().Start(c => c.TestParallelRequest).Wait(); Thread.Sleep(1000); - var emitter = container.GetInstance(); - emitter.Run().Wait(); + //var emitter = container.GetInstance(); + //emitter.Run().Wait(); } } - catch (Exception e) + //catch (Exception e) { - Console.WriteLine(e.ToString()); - Console.ReadKey(); + // Console.WriteLine(e.ToString()); + // Console.ReadKey(); } } } diff --git a/Test/Test.csproj b/Test/Test.csproj index 7aa20f6..00f9c1e 100644 --- a/Test/Test.csproj +++ b/Test/Test.csproj @@ -11,6 +11,7 @@ + diff --git a/appveyor.yml b/appveyor.yml index d2d5812..8a442c8 100644 --- a/appveyor.yml +++ b/appveyor.yml @@ -20,6 +20,8 @@ after_build: - cmd: appveyor PushArtifact "Tapeti.Flow.%GitVersion_NuGetVersion%.nupkg" - cmd: nuget pack Tapeti.Flow.SQL\Tapeti.Flow.SQL.nuspec -version "%GitVersion_NuGetVersion%" -prop "target=%CONFIGURATION%" - cmd: appveyor PushArtifact "Tapeti.Flow.SQL.%GitVersion_NuGetVersion%.nupkg" + - cmd: nuget pack Tapeti.Transient\Tapeti.Transient.nuspec -version "%GitVersion_NuGetVersion%" -prop "target=%CONFIGURATION%" + - cmd: appveyor PushArtifact "Tapeti.Transient.%GitVersion_NuGetVersion%.nupkg" - cmd: nuget pack Tapeti.SimpleInjector\Tapeti.SimpleInjector.nuspec -version "%GitVersion_NuGetVersion%" -prop "target=%CONFIGURATION%" - cmd: appveyor PushArtifact "Tapeti.SimpleInjector.%GitVersion_NuGetVersion%.nupkg" - cmd: nuget pack Tapeti.Serilog\Tapeti.Serilog.nuspec -version "%GitVersion_NuGetVersion%" -prop "target=%CONFIGURATION%"