diff --git a/Examples/01-PublishSubscribe/01-PublishSubscribe.csproj b/Examples/01-PublishSubscribe/01-PublishSubscribe.csproj index 2b83fe4..707f649 100644 --- a/Examples/01-PublishSubscribe/01-PublishSubscribe.csproj +++ b/Examples/01-PublishSubscribe/01-PublishSubscribe.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp2.1 + net5.0 _01_PublishSubscribe diff --git a/Examples/02-DeclareDurableQueues/02-DeclareDurableQueues.csproj b/Examples/02-DeclareDurableQueues/02-DeclareDurableQueues.csproj index 8a580e7..6049662 100644 --- a/Examples/02-DeclareDurableQueues/02-DeclareDurableQueues.csproj +++ b/Examples/02-DeclareDurableQueues/02-DeclareDurableQueues.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp2.1 + net5.0 _02_DeclareDurableQueues diff --git a/Examples/03-FlowRequestResponse/03-FlowRequestResponse.csproj b/Examples/03-FlowRequestResponse/03-FlowRequestResponse.csproj index 504511e..3af6a96 100644 --- a/Examples/03-FlowRequestResponse/03-FlowRequestResponse.csproj +++ b/Examples/03-FlowRequestResponse/03-FlowRequestResponse.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp2.1 + net5.0 _03_FlowRequestResponse diff --git a/Examples/04-Transient/04-Transient.csproj b/Examples/04-Transient/04-Transient.csproj index 93269e8..7e1a085 100644 --- a/Examples/04-Transient/04-Transient.csproj +++ b/Examples/04-Transient/04-Transient.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp2.1 + net5.0 _04_Transient diff --git a/Examples/05-SpeedTest/05-SpeedTest.csproj b/Examples/05-SpeedTest/05-SpeedTest.csproj index 7ef9e09..e51609f 100644 --- a/Examples/05-SpeedTest/05-SpeedTest.csproj +++ b/Examples/05-SpeedTest/05-SpeedTest.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp2.1 + net5.0 _05_SpeedTest diff --git a/Examples/06-StatelessRequestResponse/06-StatelessRequestResponse.csproj b/Examples/06-StatelessRequestResponse/06-StatelessRequestResponse.csproj index 62a3961..347bb27 100644 --- a/Examples/06-StatelessRequestResponse/06-StatelessRequestResponse.csproj +++ b/Examples/06-StatelessRequestResponse/06-StatelessRequestResponse.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp2.1 + net5.0 _06_StatelessRequestResponse diff --git a/Examples/07-ParallelizationTest/07-ParallelizationTest.csproj b/Examples/07-ParallelizationTest/07-ParallelizationTest.csproj index ecd2513..a51a70e 100644 --- a/Examples/07-ParallelizationTest/07-ParallelizationTest.csproj +++ b/Examples/07-ParallelizationTest/07-ParallelizationTest.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp3.1 + net5.0 _07_ParallelizationTest diff --git a/Examples/08-MessageHandlerLogging/08-MessageHandlerLogging.csproj b/Examples/08-MessageHandlerLogging/08-MessageHandlerLogging.csproj index 12b218e..c249120 100644 --- a/Examples/08-MessageHandlerLogging/08-MessageHandlerLogging.csproj +++ b/Examples/08-MessageHandlerLogging/08-MessageHandlerLogging.csproj @@ -2,7 +2,7 @@ Exe - netcoreapp3.1 + net5.0 _08_MessageHandlerLogging diff --git a/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs b/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs index 19f7a8b..3da64b3 100644 --- a/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs +++ b/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs @@ -37,24 +37,25 @@ namespace Tapeti.Flow.SQL /// - public async Task>> GetStates() + public async Task>> GetStates() { return await SqlRetryHelper.Execute(async () => { using (var connection = await GetConnection()) { - var flowQuery = new SqlCommand($"select FlowID, StateJson from {tableName}", connection); + var flowQuery = new SqlCommand($"select FlowID, CreationTime, StateJson from {tableName}", connection); var flowReader = await flowQuery.ExecuteReaderAsync(); - var result = new List>(); + var result = new List>(); while (await flowReader.ReadAsync()) { var flowID = flowReader.GetGuid(0); - var stateJson = flowReader.GetString(1); + var creationTime = flowReader.GetDateTime(1); + var stateJson = flowReader.GetString(2); var state = JsonConvert.DeserializeObject(stateJson); - result.Add(new KeyValuePair(flowID, state)); + result.Add(new FlowRecord(flowID, creationTime, state)); } return result; diff --git a/Tapeti.Flow/Default/FlowStore.cs b/Tapeti.Flow/Default/FlowStore.cs index 8b0aa96..044e9d0 100644 --- a/Tapeti.Flow/Default/FlowStore.cs +++ b/Tapeti.Flow/Default/FlowStore.cs @@ -18,11 +18,13 @@ namespace Tapeti.Flow.Default private class CachedFlowState { public readonly FlowState FlowState; + public readonly DateTime CreationTime; public readonly bool IsPersistent; - public CachedFlowState(FlowState flowState, bool isPersistent) + public CachedFlowState(FlowState flowState, DateTime creationTime, bool isPersistent) { FlowState = flowState; + CreationTime = creationTime; IsPersistent = isPersistent; } } @@ -64,12 +66,12 @@ namespace Tapeti.Flow.Default { foreach (var flowStateRecord in await repository.GetStates()) { - flowStates.TryAdd(flowStateRecord.Key, new CachedFlowState(flowStateRecord.Value, true)); + flowStates.TryAdd(flowStateRecord.FlowID, new CachedFlowState(flowStateRecord.FlowState, flowStateRecord.CreationTime, true)); - foreach (var continuation in flowStateRecord.Value.Continuations) + foreach (var continuation in flowStateRecord.FlowState.Continuations) { - ValidateContinuation(flowStateRecord.Key, continuation.Key, continuation.Value); - continuationLookup.GetOrAdd(continuation.Key, flowStateRecord.Key); + ValidateContinuation(flowStateRecord.FlowID, continuation.Key, continuation.Value); + continuationLookup.GetOrAdd(continuation.Key, flowStateRecord.FlowID); } } } @@ -134,6 +136,18 @@ namespace Tapeti.Flow.Default } + /// + public Task> GetActiveFlows(TimeSpan minimumAge) + { + var maximumDateTime = DateTime.UtcNow - minimumAge; + + return Task.FromResult(flowStates + .Where(p => p.Value.CreationTime <= maximumDateTime) + .Select(p => new ActiveFlow(p.Key, p.Value.CreationTime)) + .ToArray() as IEnumerable); + } + + private class FlowStateLock : IFlowStateLock { private readonly FlowStore owner; @@ -190,7 +204,7 @@ namespace Tapeti.Flow.Default var isNew = cachedFlowState == null; var wasPersistent = cachedFlowState?.IsPersistent ?? false; - cachedFlowState = new CachedFlowState(newFlowState, persistent); + cachedFlowState = new CachedFlowState(newFlowState, isNew ? DateTime.UtcNow : cachedFlowState.CreationTime, persistent); owner.flowStates[FlowID] = cachedFlowState; if (persistent) @@ -198,8 +212,7 @@ namespace Tapeti.Flow.Default // Storing the flowstate in the underlying repository if (isNew) { - var now = DateTime.UtcNow; - await owner.repository.CreateState(FlowID, cachedFlowState.FlowState, now); + await owner.repository.CreateState(FlowID, cachedFlowState.FlowState, cachedFlowState.CreationTime); } else { diff --git a/Tapeti.Flow/Default/NonPersistentFlowRepository.cs b/Tapeti.Flow/Default/NonPersistentFlowRepository.cs index b20bfd6..b1aa283 100644 --- a/Tapeti.Flow/Default/NonPersistentFlowRepository.cs +++ b/Tapeti.Flow/Default/NonPersistentFlowRepository.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading.Tasks; namespace Tapeti.Flow.Default @@ -10,9 +11,9 @@ namespace Tapeti.Flow.Default /// public class NonPersistentFlowRepository : IFlowRepository { - Task>> IFlowRepository.GetStates() + Task>> IFlowRepository.GetStates() { - return Task.FromResult(new List>()); + return Task.FromResult(Enumerable.Empty>()); } /// diff --git a/Tapeti.Flow/IFlowRepository.cs b/Tapeti.Flow/IFlowRepository.cs index f147684..cde801c 100644 --- a/Tapeti.Flow/IFlowRepository.cs +++ b/Tapeti.Flow/IFlowRepository.cs @@ -13,30 +13,64 @@ namespace Tapeti.Flow /// Load the previously persisted flow states. /// /// A list of flow states, where the key is the unique Flow ID and the value is the deserialized T. - Task>> GetStates(); + Task>> GetStates(); /// /// Stores a new flow state. Guaranteed to be run in a lock for the specified flow ID. /// - /// - /// - /// + /// The unique ID of the flow. + /// The flow state to be stored. + /// The time when the flow was initially created. /// Task CreateState(Guid flowID, T state, DateTime timestamp); /// /// Updates an existing flow state. Guaranteed to be run in a lock for the specified flow ID. /// - /// - /// - /// - /// + /// The unique ID of the flow. + /// The flow state to be stored. Task UpdateState(Guid flowID, T state); /// /// Delete a flow state. Guaranteed to be run in a lock for the specified flow ID. /// - /// + /// The unique ID of the flow. Task DeleteState(Guid flowID); } + + + /// + /// Contains information about a persisted flow state. + /// + public class FlowRecord + { + /// + /// The unique ID of the flow. + /// + public Guid FlowID { get; } + + /// + /// The time when the flow was initially created. + /// + public DateTime CreationTime { get; } + + /// + /// The stored flow state. + /// + public T FlowState { get; } + + + /// + /// Creates a new instance of a FlowRecord. + /// + /// + /// + /// + public FlowRecord(Guid flowID, DateTime creationTime, T flowState) + { + FlowID = flowID; + CreationTime = creationTime; + FlowState = flowState; + } + } } diff --git a/Tapeti.Flow/IFlowStore.cs b/Tapeti.Flow/IFlowStore.cs index 21c4337..b3720b3 100644 --- a/Tapeti.Flow/IFlowStore.cs +++ b/Tapeti.Flow/IFlowStore.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading.Tasks; using Tapeti.Flow.Default; @@ -29,6 +30,15 @@ namespace Tapeti.Flow /// /// Task LockFlowState(Guid flowID); + + /// + /// Returns information about the currently active flows. + /// + /// + /// This is intended for monitoring purposes and should be treated as a snapshot. + /// + /// The minimum age of the flow before it is included in the result. Set to TimeSpan.Zero to return all active flows. + Task> GetActiveFlows(TimeSpan minimumAge); } @@ -60,4 +70,33 @@ namespace Tapeti.Flow /// Task DeleteFlowState(); } + + + /// + /// Contains information about an active flow, as returned by . + /// + public class ActiveFlow + { + /// + /// The ID of the active flow. + /// + public Guid FlowID { get; } + + /// + /// The time when the flow was initially created. + /// + public DateTime CreationTime { get; } + + + /// + /// Create a new instance of an ActiveFlow. + /// + /// The ID of the active flow. + /// The time when the flow was initially created. + public ActiveFlow(Guid flowID, DateTime creationTime) + { + FlowID = flowID; + CreationTime = creationTime; + } + } } diff --git a/Tapeti/Config/IMessageContext.cs b/Tapeti/Config/IMessageContext.cs index 7d2db6d..6fea4cc 100644 --- a/Tapeti/Config/IMessageContext.cs +++ b/Tapeti/Config/IMessageContext.cs @@ -1,4 +1,5 @@ using System; +using System.Threading; // ReSharper disable UnusedMemberInSuper.Global - public API // ReSharper disable UnusedMember.Global @@ -50,6 +51,13 @@ namespace Tapeti.Config /// IBinding Binding { get; } + /// + /// Contains a CancellationToken which is cancelled when the connection to the RabbitMQ server is closed. + /// Note that this token is cancelled regardless of whether the connection will be reestablished, as any + /// messages still in the queue will be redelivered with a new token. + /// + CancellationToken ConnectionClosed { get; } + /// /// Stores additional properties in the message context which can be passed between middleware stages. /// diff --git a/Tapeti/Connection/TapetiConsumer.cs b/Tapeti/Connection/TapetiConsumer.cs index 6d67241..fba63dd 100644 --- a/Tapeti/Connection/TapetiConsumer.cs +++ b/Tapeti/Connection/TapetiConsumer.cs @@ -74,7 +74,8 @@ namespace Tapeti.Connection RawBody = body, Message = message, Properties = properties, - Binding = null + Binding = null, + ConnectionClosed = CancellationToken.None }; var exceptionContext = new ExceptionStrategyContext(emptyContext, dispatchException); @@ -118,7 +119,8 @@ namespace Tapeti.Connection RawBody = messageContextData.RawBody, Message = message, Properties = messageContextData.Properties, - Binding = binding + Binding = binding, + ConnectionClosed = cancellationToken }; try diff --git a/Tapeti/Default/CancellationTokenBinding.cs b/Tapeti/Default/CancellationTokenBinding.cs new file mode 100644 index 0000000..01c8a72 --- /dev/null +++ b/Tapeti/Default/CancellationTokenBinding.cs @@ -0,0 +1,25 @@ +using System; +using System.Linq; +using System.Threading; +using Tapeti.Config; + +namespace Tapeti.Default +{ + /// + /// + /// Binds a parameter of type CancellationToken to a token which is cancelled when the RabbitMQ connection is closed. + /// Similar to and very much inspired by ASP.NET's RequestAborted CancellationToken. + /// This middleware is included by default in the standard TapetiConfig. + /// + public class CancellationTokenBinding : IControllerBindingMiddleware + { + /// + public void Handle(IControllerBindingContext context, Action next) + { + foreach (var parameter in context.Parameters.Where(p => !p.HasBinding && p.Info.ParameterType == typeof(CancellationToken))) + parameter.SetBinding(messageContext => messageContext.ConnectionClosed); + + next(); + } + } +} diff --git a/Tapeti/Default/MessageContext.cs b/Tapeti/Default/MessageContext.cs index 40a153c..3b72e77 100644 --- a/Tapeti/Default/MessageContext.cs +++ b/Tapeti/Default/MessageContext.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; using Tapeti.Config; @@ -34,7 +35,10 @@ namespace Tapeti.Default /// public IBinding Binding { get; set; } - + /// + public CancellationToken ConnectionClosed { get; set; } + + public void Store(T payload) where T : IMessageContextPayload { payloads.Add(typeof(T), payload); diff --git a/Tapeti/TapetiConfig.cs b/Tapeti/TapetiConfig.cs index 57abd6e..ec2234c 100644 --- a/Tapeti/TapetiConfig.cs +++ b/Tapeti/TapetiConfig.cs @@ -35,6 +35,7 @@ namespace Tapeti Use(new DependencyResolverBinding()); Use(new PublishResultBinding()); + Use(new CancellationTokenBinding()); // Registered last so it runs first and the MessageClass is known to other middleware Use(new MessageBinding()); diff --git a/Tapeti/TapetiConnection.cs b/Tapeti/TapetiConnection.cs index f9cb6e5..f26e3ca 100644 --- a/Tapeti/TapetiConnection.cs +++ b/Tapeti/TapetiConnection.cs @@ -151,10 +151,8 @@ namespace Tapeti protected virtual void OnConnected(ConnectedEventArgs e) { var connectedEvent = Connected; - if (connectedEvent == null) - return; - - Task.Run(() => connectedEvent.Invoke(this, e)); + if (connectedEvent != null) + Task.Run(() => connectedEvent.Invoke(this, e)); } /// @@ -162,13 +160,11 @@ namespace Tapeti /// protected virtual void OnReconnected(ConnectedEventArgs e) { - var reconnectedEvent = Reconnected; - if (reconnectedEvent == null && subscriber == null) - return; - subscriber?.Reconnect(); - Task.Run(() => reconnectedEvent?.Invoke(this, e)); + var reconnectedEvent = Reconnected; + if (reconnectedEvent != null) + Task.Run(() => reconnectedEvent?.Invoke(this, e)); } /// @@ -176,13 +172,11 @@ namespace Tapeti /// protected virtual void OnDisconnected(DisconnectedEventArgs e) { - var disconnectedEvent = Disconnected; - if (disconnectedEvent == null) - return; - subscriber?.Disconnect(); - Task.Run(() => disconnectedEvent.Invoke(this, e)); + var disconnectedEvent = Disconnected; + if (disconnectedEvent != null) + Task.Run(() => disconnectedEvent.Invoke(this, e)); } } } diff --git a/docs/compatibility.rst b/docs/compatibility.rst new file mode 100644 index 0000000..0fac765 --- /dev/null +++ b/docs/compatibility.rst @@ -0,0 +1,66 @@ +Compatibility +============= + +ASP.NET Core +------------ +When integrating Tapeti into an ASP.NET Core service, depending on your naming conventions you may run into an issue where ASP.NET tries to register all your messaging controllers as API controllers. This is because by default any class ending in "Controller" will be picked up by ASP.NET. + +You can rename your Tapeti controller classes as long as there are no persisted Tapeti Flows for that controller. + +Alternatively, you can filter the classes that ASP.NET will register by using a custom ControllerFeatureProvider. + + +Whitelist ASP.NET controllers +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +If all your ASP.NET controllers use the [ApiController] attribute, you can simply whitelist classes which are annotated with that attribute, ensuring the class name is not relevant to ASP.NET: + +:: + + public class APIControllersOnlyFeatureProvider : ControllerFeatureProvider + { + protected override bool IsController(TypeInfo typeInfo) + { + return typeInfo.IsClass && typeInfo.IsPublic && !typeInfo.IsAbstract && + typeInfo.GetCustomAttribute() != null; + } + } + + +Blacklist Tapeti controllers +^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +If instead you want the default ASP.NET behaviour but only exclude Tapeti messaging controllers, you can decorate the default feature provider as follows: + +:: + + public class ExcludeMessagingControllerFeatureProvider : ControllerFeatureProvider + { + protected override bool IsController(TypeInfo typeInfo) + { + return base.IsController(typeInfo) && + typeInfo.GetCustomAttribute() == null; + } + } + + +Replace feature provider +^^^^^^^^^^^^^^^^^^^^^^^^ +In either case, replace the default ControllerFeatureProvider in the ConfigureServices method of your startup class: + +:: + + public void ConfigureServices(IServiceCollection services) + { + services + .AddControllers() + .ConfigureApplicationPartManager(manager => + { + var controllerFeatureProvider = manager.FeatureProviders + .FirstOrDefault(provider => provider is ControllerFeatureProvider); + + if (controllerFeatureProvider != null) + manager.FeatureProviders.Remove(controllerFeatureProvider); + + manager.FeatureProviders.Add(new APIControllersOnlyFeatureProvider()); + // or: manager.FeatureProviders.Add(new ExcludeMessagingControllerFeatureProvider()); + }); + } \ No newline at end of file diff --git a/docs/flow.rst b/docs/flow.rst index 25cd90f..53bcabb 100644 --- a/docs/flow.rst +++ b/docs/flow.rst @@ -289,3 +289,6 @@ Then install the Tapeti.Flow.SQL NuGet package and register the SqlConnectionFlo .WithFlow() .RegisterAllControllers() .Build(); + + +.. caution:: The controller and method names for response handlers and converge methods are stored in the flow and must be valid when they are loaded again. Keep that in mind if you want to refactor the code; either keep the original class and method temporarily for backwards compatibility, optionally redirecting them internally to the new code, or make sure there are no persisted flows remaining. \ No newline at end of file diff --git a/docs/gettingstarted.rst b/docs/gettingstarted.rst index a4826c7..98dbeb3 100644 --- a/docs/gettingstarted.rst +++ b/docs/gettingstarted.rst @@ -74,7 +74,7 @@ A message is simply a plain object which can be serialized using `Json.NET or a message class. -The name of the method is not important to Tapeti. Any parameter other than the first will be resolved using the IoC container, although it is considered best practice to use the constructor for dependency injection instead. +The name of the method is not important to Tapeti. Any parameter other than the first will be resolved in two ways: + +1. Registered middleware can alter the behaviour of parameters. Tapeti includes one by default for CancellationToken parameters, see :ref:`parameterbinding` in :doc:`indepth`. +2. Any remaining parameters are resolved using the IoC container, although it is considered best practice to use the constructor for dependency injection instead. A new controller is instantiated for each message, so it is safe to use public or private fields to store state while handling the message. Just don't expect it to be there for the next message. If you need this behaviour, take a look at the :doc:`flow`! diff --git a/docs/indepth.rst b/docs/indepth.rst index c4e4fe6..8b68d6b 100644 --- a/docs/indepth.rst +++ b/docs/indepth.rst @@ -9,6 +9,46 @@ As described in the Getting started guide, a message is a plain object which can When communicating between services it is considered best practice to define messages in separate class library assemblies which can be referenced in other services. This establishes a public interface between services and components without binding to the implementation. +.. _parameterbinding: + +Parameter binding +----------------- +Tapeti will bind the parameters of message handler methods using the registered binding middleware. + +Although stated in the Getting started guide that the first parameter is always assumed to be the message class, this is in fact handled by one of the default binding middleware implementations instead of being hardcoded in Tapeti. All of the default implementations play nice and will only apply to parameters not already bound by other middleware, making it easy to extend or change the default behaviour if desired. + +In addition to the message class parameter, two additional default implementations are included: + + +CancellationToken +^^^^^^^^^^^^^^^^^ +Similar to ASP.NET, Tapeti will bind parameters of type CancellationToken to a token which is cancelled when the connection to the RabbitMQ server is closed. + +.. note:: This does not indicate whether the connection was closed by the application or lost unexpectedly, either scenario will cancel the token. This is by design, as any message in-flight will be put back on the queue and redelivered anyways. + +Internally this CancellationToken is called ConnectionClosed, but any name can be used. For example: + +:: + + public async Task CountRabbits(CountRequestMessage message, + CancellationToken cancellationToken) + { + var count = await rabbitRepository.Count(cancellationToken); + + return new CountRabbitsResponseMessage + { + Count = count + }; + } + + +Dependency injection +^^^^^^^^^^^^^^^^^^^^ +Any parameter not bound by any other means will be resolved using the IoC container which is passed to the TapetiConnection. + +.. note:: It is considered best practice to use the constructor for dependency injection instead. + + Enums ----- Special care must be taken when using enums in messages. For example, you have several services consuming a message containing an enum field. Some services will have logic which depends on a specific value, others will not use that specific field at all. @@ -82,6 +122,7 @@ If all message handlers bound to a durable queue are marked as obsolete, includi If there are still messages in the queue it's pending removal will be logged but the consumers will run as normal to empty the queue. The queue will then remain until it is checked again when the application is restarted. + Request - response ------------------ Messages can be annotated with the Request attribute to indicate that they require a response. For example: diff --git a/docs/index.rst b/docs/index.rst index cbc0898..c88d375 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -7,6 +7,7 @@ Tapeti documentation introduction gettingstarted + compatibility indepth dataannotations flow