From 165680fd388d59ee25f70a96a0457da75832b665 Mon Sep 17 00:00:00 2001 From: Mark van Renswoude Date: Wed, 9 Feb 2022 11:26:56 +0100 Subject: [PATCH] Added ValueTask support - This is a breaking change for custom middleware implementations Added validation for return type handling - This may be breaking for incorrect implementations, but highly unlikely --- .../ReceivingMessageController.cs | 4 +- .../05-SpeedTest/SpeedMessageController.cs | 2 + .../ReceivingMessageController.cs | 4 +- .../DataAnnotationsMessageMiddleware.cs | 4 +- .../DataAnnotationsPublishMiddleware.cs | 4 +- .../SqlConnectionFlowRepository.cs | 12 ++-- Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj | 5 ++ Tapeti.Flow/Default/FlowBindingMiddleware.cs | 50 ++++++++++---- Tapeti.Flow/Default/FlowContext.cs | 10 ++- .../Default/FlowContinuationMiddleware.cs | 8 +-- Tapeti.Flow/Default/FlowProvider.cs | 4 +- Tapeti.Flow/Default/FlowStore.cs | 22 +++---- .../Default/NonPersistentFlowRepository.cs | 16 ++--- Tapeti.Flow/IFlowProvider.cs | 4 +- Tapeti.Flow/IFlowRepository.cs | 8 +-- Tapeti.Flow/IFlowStore.cs | 14 ++-- Tapeti.Flow/Tapeti.Flow.csproj | 4 +- .../MessageHandlerLoggingMessageMiddleware.cs | 3 +- Tapeti/Config/IControllerBindingContext.cs | 2 +- Tapeti/Config/IControllerCleanupMiddleware.cs | 2 +- Tapeti/Config/IControllerFilterMiddleware.cs | 2 +- Tapeti/Config/IControllerMessageMiddleware.cs | 2 +- Tapeti/Config/IMessageMiddleware.cs | 2 +- Tapeti/Config/IPublishMiddleware.cs | 2 +- Tapeti/Connection/TapetiSubscriber.cs | 3 +- Tapeti/Default/ControllerMethodBinding.cs | 41 ++++-------- Tapeti/Default/PublishResultBinding.cs | 51 ++++++++++----- Tapeti/Helpers/MiddlewareHelper.cs | 4 +- Tapeti/Helpers/TaskTypeHelper.cs | 65 +++++++++++++++---- Tapeti/Tapeti.csproj | 1 + Tapeti/TapetiConfigControllers.cs | 2 +- 31 files changed, 215 insertions(+), 142 deletions(-) diff --git a/Examples/03-FlowRequestResponse/ReceivingMessageController.cs b/Examples/03-FlowRequestResponse/ReceivingMessageController.cs index f9c83d5..97c498c 100644 --- a/Examples/03-FlowRequestResponse/ReceivingMessageController.cs +++ b/Examples/03-FlowRequestResponse/ReceivingMessageController.cs @@ -9,8 +9,7 @@ namespace _03_FlowRequestResponse public class ReceivingMessageController { // No publisher required, responses can simply be returned - #pragma warning disable CA1822 // Mark members as static - not supported yet by Tapeti - public async Task HandleQuoteRequest(QuoteRequestMessage message) + public static async Task HandleQuoteRequest(QuoteRequestMessage message) { var quote = message.Amount switch { @@ -29,6 +28,5 @@ namespace _03_FlowRequestResponse Quote = quote }; } - #pragma warning restore CA1822 } } diff --git a/Examples/05-SpeedTest/SpeedMessageController.cs b/Examples/05-SpeedTest/SpeedMessageController.cs index 83af176..b0e5386 100644 --- a/Examples/05-SpeedTest/SpeedMessageController.cs +++ b/Examples/05-SpeedTest/SpeedMessageController.cs @@ -15,9 +15,11 @@ namespace _05_SpeedTest } + #pragma warning disable IDE0060 // Remove unused parameter public void HandleSpeedTestMessage(SpeedTestMessage message) { messageCounter.Add(); } + #pragma warning restore IDE0060 } } diff --git a/Examples/06-StatelessRequestResponse/ReceivingMessageController.cs b/Examples/06-StatelessRequestResponse/ReceivingMessageController.cs index 9133713..4a2704b 100644 --- a/Examples/06-StatelessRequestResponse/ReceivingMessageController.cs +++ b/Examples/06-StatelessRequestResponse/ReceivingMessageController.cs @@ -8,8 +8,7 @@ namespace _06_StatelessRequestResponse public class ReceivingMessageController { // No publisher required, responses can simply be returned - #pragma warning disable CA1822 // Mark members as static - not supported yet by Tapeti - public QuoteResponseMessage HandleQuoteRequest(QuoteRequestMessage message) + public static QuoteResponseMessage HandleQuoteRequest(QuoteRequestMessage message) { var quote = message.Amount switch { @@ -25,6 +24,5 @@ namespace _06_StatelessRequestResponse Quote = quote }; } - #pragma warning restore CA1822 } } diff --git a/Tapeti.DataAnnotations/DataAnnotationsMessageMiddleware.cs b/Tapeti.DataAnnotations/DataAnnotationsMessageMiddleware.cs index 8b2ed85..d2df50a 100644 --- a/Tapeti.DataAnnotations/DataAnnotationsMessageMiddleware.cs +++ b/Tapeti.DataAnnotations/DataAnnotationsMessageMiddleware.cs @@ -12,12 +12,12 @@ namespace Tapeti.DataAnnotations internal class DataAnnotationsMessageMiddleware : IMessageMiddleware { /// - public async Task Handle(IMessageContext context, Func next) + public ValueTask Handle(IMessageContext context, Func next) { var validationContext = new ValidationContext(context.Message); Validator.ValidateObject(context.Message, validationContext, true); - await next(); + return next(); } } } diff --git a/Tapeti.DataAnnotations/DataAnnotationsPublishMiddleware.cs b/Tapeti.DataAnnotations/DataAnnotationsPublishMiddleware.cs index 514989c..7908cda 100644 --- a/Tapeti.DataAnnotations/DataAnnotationsPublishMiddleware.cs +++ b/Tapeti.DataAnnotations/DataAnnotationsPublishMiddleware.cs @@ -12,12 +12,12 @@ namespace Tapeti.DataAnnotations internal class DataAnnotationsPublishMiddleware : IPublishMiddleware { /// - public async Task Handle(IPublishContext context, Func next) + public ValueTask Handle(IPublishContext context, Func next) { var validationContext = new ValidationContext(context.Message); Validator.ValidateObject(context.Message, validationContext, true); - await next(); + return next(); } } } diff --git a/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs b/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs index 3da64b3..bdf3979 100644 --- a/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs +++ b/Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs @@ -5,6 +5,10 @@ using System.Data.SqlClient; using System.Threading.Tasks; using Newtonsoft.Json; +// Neither of these are available in language version 7 required for .NET Standard 2.0 +// ReSharper disable ConvertToUsingDeclaration +// ReSharper disable UseAwaitUsing + namespace Tapeti.Flow.SQL { /// @@ -37,7 +41,7 @@ namespace Tapeti.Flow.SQL /// - public async Task>> GetStates() + public async ValueTask>> GetStates() { return await SqlRetryHelper.Execute(async () => { @@ -64,7 +68,7 @@ namespace Tapeti.Flow.SQL } /// - public async Task CreateState(Guid flowID, T state, DateTime timestamp) + public async ValueTask CreateState(Guid flowID, T state, DateTime timestamp) { await SqlRetryHelper.Execute(async () => { @@ -88,7 +92,7 @@ namespace Tapeti.Flow.SQL } /// - public async Task UpdateState(Guid flowID, T state) + public async ValueTask UpdateState(Guid flowID, T state) { await SqlRetryHelper.Execute(async () => { @@ -108,7 +112,7 @@ namespace Tapeti.Flow.SQL } /// - public async Task DeleteState(Guid flowID) + public async ValueTask DeleteState(Guid flowID) { await SqlRetryHelper.Execute(async () => { diff --git a/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj b/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj index e4be9ef..deaa8ce 100644 --- a/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj +++ b/Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj @@ -17,6 +17,11 @@ 1701;1702 + + + IDE0063 + + diff --git a/Tapeti.Flow/Default/FlowBindingMiddleware.cs b/Tapeti.Flow/Default/FlowBindingMiddleware.cs index c0f9565..83564c8 100644 --- a/Tapeti.Flow/Default/FlowBindingMiddleware.cs +++ b/Tapeti.Flow/Default/FlowBindingMiddleware.cs @@ -31,6 +31,9 @@ namespace Tapeti.Flow.Default if (continuationAttribute == null) return; + if (context.Method.IsStatic) + throw new ArgumentException($"Continuation attribute is not valid on static methods in controller {context.Method.DeclaringType?.FullName}, method {context.Method.Name}"); + context.SetBindingTargetMode(BindingTargetMode.Direct); context.Use(new FlowContinuationMiddleware()); @@ -52,7 +55,7 @@ namespace Tapeti.Flow.Default context.Result.SetHandler((messageContext, value) => HandleParallelResponse(messageContext)); } else - throw new ArgumentException($"Result type must be IYieldPoint, Task or void in controller {context. Method.DeclaringType?.FullName}, method {context.Method.Name}"); + throw new ArgumentException($"Result type must be IYieldPoint, Task or void in controller {context.Method.DeclaringType?.FullName}, method {context.Method.Name}"); foreach (var parameter in context.Parameters.Where(p => !p.HasBinding && p.Info.ParameterType == typeof(IFlowParallelRequest))) @@ -62,34 +65,53 @@ namespace Tapeti.Flow.Default private static void RegisterYieldPointResult(IControllerBindingContext context) { - if (!context.Result.Info.ParameterType.IsTypeOrTaskOf(typeof(IYieldPoint), out var isTaskOf)) + if (!context.Result.Info.ParameterType.IsTypeOrTaskOf(typeof(IYieldPoint), out var taskType)) return; - if (isTaskOf) + if (context.Method.IsStatic) + throw new ArgumentException($"Yield points are not valid on static methods in controller {context.Method.DeclaringType?.FullName}, method {context.Method.Name}"); + + switch (taskType) { - context.Result.SetHandler(async (messageContext, value) => - { - var yieldPoint = await (Task)value; - if (yieldPoint != null) - await HandleYieldPoint(messageContext, yieldPoint); - }); + case TaskType.None: + context.Result.SetHandler((messageContext, value) => HandleYieldPoint(messageContext, (IYieldPoint)value)); + break; + + case TaskType.Task: + context.Result.SetHandler(async (messageContext, value) => + { + var yieldPoint = await (Task)value; + if (yieldPoint != null) + await HandleYieldPoint(messageContext, yieldPoint); + }); + break; + + case TaskType.ValueTask: + context.Result.SetHandler(async (messageContext, value) => + { + var yieldPoint = await (ValueTask)value; + if (yieldPoint != null) + await HandleYieldPoint(messageContext, yieldPoint); + }); + break; + + default: + throw new ArgumentOutOfRangeException(); } - else - context.Result.SetHandler((messageContext, value) => HandleYieldPoint(messageContext, (IYieldPoint)value)); } - private static Task HandleYieldPoint(IMessageContext context, IYieldPoint yieldPoint) + private static ValueTask HandleYieldPoint(IMessageContext context, IYieldPoint yieldPoint) { var flowHandler = context.Config.DependencyResolver.Resolve(); return flowHandler.Execute(new FlowHandlerContext(context), yieldPoint); } - private static Task HandleParallelResponse(IMessageContext context) + private static ValueTask HandleParallelResponse(IMessageContext context) { if (context.TryGet(out var flowPayload) && flowPayload.FlowIsConverging) - return Task.CompletedTask; + return default; var flowHandler = context.Config.DependencyResolver.Resolve(); return flowHandler.Execute(new FlowHandlerContext(context), new DelegateYieldPoint(async flowContext => diff --git a/Tapeti.Flow/Default/FlowContext.cs b/Tapeti.Flow/Default/FlowContext.cs index 70f98f9..8de2047 100644 --- a/Tapeti.Flow/Default/FlowContext.cs +++ b/Tapeti.Flow/Default/FlowContext.cs @@ -17,7 +17,7 @@ namespace Tapeti.Flow.Default private int deleteCalled; - public async Task Store(bool persistent) + public ValueTask Store(bool persistent) { storeCalled++; @@ -26,15 +26,13 @@ namespace Tapeti.Flow.Default if (FlowStateLock == null) throw new ArgumentNullException(nameof(FlowStateLock)); FlowState.Data = Newtonsoft.Json.JsonConvert.SerializeObject(HandlerContext.Controller); - await FlowStateLock.StoreFlowState(FlowState, persistent); + return FlowStateLock.StoreFlowState(FlowState, persistent); } - public async Task Delete() + public ValueTask Delete() { deleteCalled++; - - if (FlowStateLock != null) - await FlowStateLock.DeleteFlowState(); + return FlowStateLock?.DeleteFlowState() ?? default; } public bool IsStoredOrDeleted() diff --git a/Tapeti.Flow/Default/FlowContinuationMiddleware.cs b/Tapeti.Flow/Default/FlowContinuationMiddleware.cs index 9acbb99..f2de064 100644 --- a/Tapeti.Flow/Default/FlowContinuationMiddleware.cs +++ b/Tapeti.Flow/Default/FlowContinuationMiddleware.cs @@ -11,7 +11,7 @@ namespace Tapeti.Flow.Default /// internal class FlowContinuationMiddleware : IControllerFilterMiddleware, IControllerMessageMiddleware, IControllerCleanupMiddleware { - public async Task Filter(IMessageContext context, Func next) + public async ValueTask Filter(IMessageContext context, Func next) { if (!context.TryGet(out var controllerPayload)) return; @@ -27,7 +27,7 @@ namespace Tapeti.Flow.Default } - public async Task Handle(IMessageContext context, Func next) + public async ValueTask Handle(IMessageContext context, Func next) { if (!context.TryGet(out var controllerPayload)) return; @@ -53,7 +53,7 @@ namespace Tapeti.Flow.Default } - public async Task Cleanup(IMessageContext context, ConsumeResult consumeResult, Func next) + public async ValueTask Cleanup(IMessageContext context, ConsumeResult consumeResult, Func next) { await next(); @@ -82,7 +82,7 @@ namespace Tapeti.Flow.Default - private static async Task EnrichWithFlowContext(IMessageContext context) + private static async ValueTask EnrichWithFlowContext(IMessageContext context) { if (context.TryGet(out var flowPayload)) return flowPayload.FlowContext; diff --git a/Tapeti.Flow/Default/FlowProvider.cs b/Tapeti.Flow/Default/FlowProvider.cs index 808ccc2..394377b 100644 --- a/Tapeti.Flow/Default/FlowProvider.cs +++ b/Tapeti.Flow/Default/FlowProvider.cs @@ -197,7 +197,7 @@ namespace Tapeti.Flow.Default /// - public async Task Execute(IFlowHandlerContext context, IYieldPoint yieldPoint) + public async ValueTask Execute(IFlowHandlerContext context, IYieldPoint yieldPoint) { if (!(yieldPoint is DelegateYieldPoint executableYieldPoint)) throw new YieldPointException($"Yield point is required in controller {context.Controller.GetType().Name} for method {context.Method.Name}"); @@ -254,7 +254,7 @@ namespace Tapeti.Flow.Default /// - public Task Converge(IFlowHandlerContext context) + public ValueTask Converge(IFlowHandlerContext context) { return Execute(context, new DelegateYieldPoint(flowContext => Converge(flowContext, flowContext.ContinuationMetadata.ConvergeMethodName, flowContext.ContinuationMetadata.ConvergeMethodSync))); diff --git a/Tapeti.Flow/Default/FlowStore.cs b/Tapeti.Flow/Default/FlowStore.cs index 044e9d0..42a99e5 100644 --- a/Tapeti.Flow/Default/FlowStore.cs +++ b/Tapeti.Flow/Default/FlowStore.cs @@ -51,7 +51,7 @@ namespace Tapeti.Flow.Default /// - public async Task Load() + public async ValueTask Load() { if (inUse) throw new InvalidOperationException("Can only load the saved state once."); @@ -114,17 +114,17 @@ namespace Tapeti.Flow.Default /// - public Task FindFlowID(Guid continuationID) + public ValueTask 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); + return new ValueTask(continuationLookup.TryGetValue(continuationID, out var result) ? result : (Guid?)null); } /// - public async Task LockFlowState(Guid flowID) + public async ValueTask LockFlowState(Guid flowID) { if (!loaded) throw new InvalidOperationException("Flow store should be loaded before storing flows."); @@ -137,14 +137,14 @@ namespace Tapeti.Flow.Default /// - public Task> GetActiveFlows(TimeSpan minimumAge) + public ValueTask> GetActiveFlows(TimeSpan minimumAge) { var maximumDateTime = DateTime.UtcNow - minimumAge; - return Task.FromResult(flowStates + return new ValueTask>(flowStates .Where(p => p.Value.CreationTime <= maximumDateTime) .Select(p => new ActiveFlow(p.Key, p.Value.CreationTime)) - .ToArray() as IEnumerable); + .ToArray()); } @@ -173,15 +173,15 @@ namespace Tapeti.Flow.Default l?.Dispose(); } - public Task GetFlowState() + public ValueTask GetFlowState() { if (flowLock == null) throw new ObjectDisposedException("FlowStateLock"); - return Task.FromResult(cachedFlowState?.FlowState?.Clone()); + return new ValueTask(cachedFlowState?.FlowState?.Clone()); } - public async Task StoreFlowState(FlowState newFlowState, bool persistent) + public async ValueTask StoreFlowState(FlowState newFlowState, bool persistent) { if (flowLock == null) throw new ObjectDisposedException("FlowStateLock"); @@ -227,7 +227,7 @@ namespace Tapeti.Flow.Default } } - public async Task DeleteFlowState() + public async ValueTask DeleteFlowState() { if (flowLock == null) throw new ObjectDisposedException("FlowStateLock"); diff --git a/Tapeti.Flow/Default/NonPersistentFlowRepository.cs b/Tapeti.Flow/Default/NonPersistentFlowRepository.cs index b1aa283..bbaa40d 100644 --- a/Tapeti.Flow/Default/NonPersistentFlowRepository.cs +++ b/Tapeti.Flow/Default/NonPersistentFlowRepository.cs @@ -11,27 +11,27 @@ namespace Tapeti.Flow.Default /// public class NonPersistentFlowRepository : IFlowRepository { - Task>> IFlowRepository.GetStates() + ValueTask>> IFlowRepository.GetStates() { - return Task.FromResult(Enumerable.Empty>()); + return new ValueTask>>(Enumerable.Empty>()); } /// - public Task CreateState(Guid flowID, T state, DateTime timestamp) + public ValueTask CreateState(Guid flowID, T state, DateTime timestamp) { - return Task.CompletedTask; + return default; } /// - public Task UpdateState(Guid flowID, T state) + public ValueTask UpdateState(Guid flowID, T state) { - return Task.CompletedTask; + return default; } /// - public Task DeleteState(Guid flowID) + public ValueTask DeleteState(Guid flowID) { - return Task.CompletedTask; + return default; } } } diff --git a/Tapeti.Flow/IFlowProvider.cs b/Tapeti.Flow/IFlowProvider.cs index df8a485..d8b7abe 100644 --- a/Tapeti.Flow/IFlowProvider.cs +++ b/Tapeti.Flow/IFlowProvider.cs @@ -108,7 +108,7 @@ namespace Tapeti.Flow /// /// /// - Task Execute(IFlowHandlerContext context, IYieldPoint yieldPoint); + ValueTask Execute(IFlowHandlerContext context, IYieldPoint yieldPoint); /// @@ -120,7 +120,7 @@ namespace Tapeti.Flow /// /// Calls the converge method for a parallel flow. /// - Task Converge(IFlowHandlerContext context); + ValueTask Converge(IFlowHandlerContext context); } diff --git a/Tapeti.Flow/IFlowRepository.cs b/Tapeti.Flow/IFlowRepository.cs index cde801c..691279b 100644 --- a/Tapeti.Flow/IFlowRepository.cs +++ b/Tapeti.Flow/IFlowRepository.cs @@ -13,7 +13,7 @@ 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(); + ValueTask>> GetStates(); /// /// Stores a new flow state. Guaranteed to be run in a lock for the specified flow ID. @@ -22,20 +22,20 @@ namespace Tapeti.Flow /// The flow state to be stored. /// The time when the flow was initially created. /// - Task CreateState(Guid flowID, T state, DateTime timestamp); + ValueTask 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); + ValueTask 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); + ValueTask DeleteState(Guid flowID); } diff --git a/Tapeti.Flow/IFlowStore.cs b/Tapeti.Flow/IFlowStore.cs index b3720b3..9766e5b 100644 --- a/Tapeti.Flow/IFlowStore.cs +++ b/Tapeti.Flow/IFlowStore.cs @@ -17,19 +17,19 @@ namespace Tapeti.Flow /// If using an IFlowRepository that requires an update (such as creating tables) make /// sure it is called before calling Load. /// - Task Load(); + ValueTask Load(); /// /// Looks up the FlowID corresponding to a ContinuationID. For internal use. /// /// - Task FindFlowID(Guid continuationID); + ValueTask FindFlowID(Guid continuationID); /// /// Acquires a lock on the flow with the specified FlowID. /// /// - Task LockFlowState(Guid flowID); + ValueTask LockFlowState(Guid flowID); /// /// Returns information about the currently active flows. @@ -38,7 +38,7 @@ namespace Tapeti.Flow /// 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); + ValueTask> GetActiveFlows(TimeSpan minimumAge); } @@ -56,19 +56,19 @@ namespace Tapeti.Flow /// /// Acquires a copy of the flow state. /// - Task GetFlowState(); + ValueTask GetFlowState(); /// /// Stores the new flow state. /// /// /// - Task StoreFlowState(FlowState flowState, bool persistent); + ValueTask StoreFlowState(FlowState flowState, bool persistent); /// /// Disposes of the flow state corresponding to this Flow ID. /// - Task DeleteFlowState(); + ValueTask DeleteFlowState(); } diff --git a/Tapeti.Flow/Tapeti.Flow.csproj b/Tapeti.Flow/Tapeti.Flow.csproj index 5f4cf71..c59e707 100644 --- a/Tapeti.Flow/Tapeti.Flow.csproj +++ b/Tapeti.Flow/Tapeti.Flow.csproj @@ -1,4 +1,4 @@ - + netstandard2.0;netstandard2.1 @@ -18,7 +18,7 @@ - + IDE0066 diff --git a/Tapeti.Serilog/Middleware/MessageHandlerLoggingMessageMiddleware.cs b/Tapeti.Serilog/Middleware/MessageHandlerLoggingMessageMiddleware.cs index c8e6e73..1043039 100644 --- a/Tapeti.Serilog/Middleware/MessageHandlerLoggingMessageMiddleware.cs +++ b/Tapeti.Serilog/Middleware/MessageHandlerLoggingMessageMiddleware.cs @@ -29,7 +29,7 @@ namespace Tapeti.Serilog.Middleware } /// - public async Task Handle(IMessageContext context, Func next) + public async ValueTask Handle(IMessageContext context, Func next) { var logger = context.Config.DependencyResolver.Resolve(); @@ -41,6 +41,7 @@ namespace Tapeti.Serilog.Middleware await next(); + stopwatch.Stop(); diff --git a/Tapeti/Config/IControllerBindingContext.cs b/Tapeti/Config/IControllerBindingContext.cs index 37fb4d4..2023e22 100644 --- a/Tapeti/Config/IControllerBindingContext.cs +++ b/Tapeti/Config/IControllerBindingContext.cs @@ -19,7 +19,7 @@ namespace Tapeti.Config /// /// /// - public delegate Task ResultHandler(IMessageContext context, object value); + public delegate ValueTask ResultHandler(IMessageContext context, object value); /// diff --git a/Tapeti/Config/IControllerCleanupMiddleware.cs b/Tapeti/Config/IControllerCleanupMiddleware.cs index 86ef003..980c4a7 100644 --- a/Tapeti/Config/IControllerCleanupMiddleware.cs +++ b/Tapeti/Config/IControllerCleanupMiddleware.cs @@ -14,6 +14,6 @@ namespace Tapeti.Config /// /// /// Always call to allow the next in the chain to clean up - Task Cleanup(IMessageContext context, ConsumeResult consumeResult, Func next); + ValueTask Cleanup(IMessageContext context, ConsumeResult consumeResult, Func next); } } diff --git a/Tapeti/Config/IControllerFilterMiddleware.cs b/Tapeti/Config/IControllerFilterMiddleware.cs index 6a30e20..a4e3a35 100644 --- a/Tapeti/Config/IControllerFilterMiddleware.cs +++ b/Tapeti/Config/IControllerFilterMiddleware.cs @@ -15,6 +15,6 @@ namespace Tapeti.Config /// /// /// - Task Filter(IMessageContext context, Func next); + ValueTask Filter(IMessageContext context, Func next); } } diff --git a/Tapeti/Config/IControllerMessageMiddleware.cs b/Tapeti/Config/IControllerMessageMiddleware.cs index c381270..e497ead 100644 --- a/Tapeti/Config/IControllerMessageMiddleware.cs +++ b/Tapeti/Config/IControllerMessageMiddleware.cs @@ -14,6 +14,6 @@ namespace Tapeti.Config /// /// /// Call to pass the message to the next handler in the chain or call the controller method - Task Handle(IMessageContext context, Func next); + ValueTask Handle(IMessageContext context, Func next); } } diff --git a/Tapeti/Config/IMessageMiddleware.cs b/Tapeti/Config/IMessageMiddleware.cs index 134b5de..228a1a6 100644 --- a/Tapeti/Config/IMessageMiddleware.cs +++ b/Tapeti/Config/IMessageMiddleware.cs @@ -13,6 +13,6 @@ namespace Tapeti.Config /// /// /// Call to pass the message to the next handler in the chain - Task Handle(IMessageContext context, Func next); + ValueTask Handle(IMessageContext context, Func next); } } diff --git a/Tapeti/Config/IPublishMiddleware.cs b/Tapeti/Config/IPublishMiddleware.cs index c8069e3..e68cf88 100644 --- a/Tapeti/Config/IPublishMiddleware.cs +++ b/Tapeti/Config/IPublishMiddleware.cs @@ -13,6 +13,6 @@ namespace Tapeti.Config /// /// /// Call to pass the message to the next handler in the chain - Task Handle(IPublishContext context, Func next); + ValueTask Handle(IPublishContext context, Func next); } } diff --git a/Tapeti/Connection/TapetiSubscriber.cs b/Tapeti/Connection/TapetiSubscriber.cs index c8133cc..35cbf75 100644 --- a/Tapeti/Connection/TapetiSubscriber.cs +++ b/Tapeti/Connection/TapetiSubscriber.cs @@ -80,14 +80,13 @@ namespace Tapeti.Connection cancellationToken = initializeCancellationTokenSource.Token; - // ReSharper disable once MethodSupportsCancellation Task.Run(async () => { await ApplyBindings(cancellationToken); if (consuming && !cancellationToken.IsCancellationRequested) await ConsumeQueues(cancellationToken); - }); + }, CancellationToken.None); } diff --git a/Tapeti/Default/ControllerMethodBinding.cs b/Tapeti/Default/ControllerMethodBinding.cs index 03b44da..3884cf8 100644 --- a/Tapeti/Default/ControllerMethodBinding.cs +++ b/Tapeti/Default/ControllerMethodBinding.cs @@ -159,7 +159,7 @@ namespace Tapeti.Default /// public async Task Invoke(IMessageContext context) { - var controller = dependencyResolver.Resolve(bindingInfo.ControllerType); + var controller = Method.IsStatic ? null : dependencyResolver.Resolve(bindingInfo.ControllerType); context.Store(new ControllerMessageContextPayload(controller, context.Binding as IControllerMethodBinding)); if (!await FilterAllowed(context)) @@ -179,7 +179,7 @@ namespace Tapeti.Default await MiddlewareHelper.GoAsync( bindingInfo.CleanupMiddleware, async (handler, next) => await handler.Cleanup(context, consumeResult, next), - () => Task.CompletedTask); + () => default); } @@ -192,14 +192,14 @@ namespace Tapeti.Default () => { allowed = true; - return Task.CompletedTask; + return default; }); return allowed; } - private delegate Task MessageHandlerFunc(IMessageContext context); + private delegate ValueTask MessageHandlerFunc(IMessageContext context); private MessageHandlerFunc WrapMethod(MethodInfo method, IEnumerable parameterFactories, ResultHandler resultHandler) @@ -213,10 +213,11 @@ namespace Tapeti.Default if (method.ReturnType == typeof(Task)) return WrapTaskMethod(method, parameterFactories); - if (method.ReturnType.IsGenericType && method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>)) - return WrapGenericTaskMethod(method, parameterFactories); + if (method.ReturnType == typeof(ValueTask)) + return WrapValueTaskMethod(method, parameterFactories); - return WrapObjectMethod(method, parameterFactories); + // Breaking change in Tapeti 2.9: PublishResultBinding or other middleware should have taken care of the return value. If not, don't silently discard it. + throw new ArgumentException($"Method {method.Name} on controller {method.DeclaringType?.FullName} returns type {method.ReturnType.FullName}, which can not be handled by Tapeti or any registered middleware"); } @@ -246,7 +247,7 @@ namespace Tapeti.Default try { method.Invoke(controllerPayload.Controller, parameterFactories.Select(p => p(context)).ToArray()); - return Task.CompletedTask; + return default; } catch (Exception e) { @@ -264,7 +265,7 @@ namespace Tapeti.Default var controllerPayload = context.Get(); try { - return (Task) method.Invoke(controllerPayload.Controller, parameterFactories.Select(p => p(context)).ToArray()); + return new ValueTask((Task) method.Invoke(controllerPayload.Controller, parameterFactories.Select(p => p(context)).ToArray())); } catch (Exception e) { @@ -275,32 +276,14 @@ namespace Tapeti.Default } - private MessageHandlerFunc WrapGenericTaskMethod(MethodBase method, IEnumerable parameterFactories) - { - return context => - { - var controllerPayload = context.Get(); - try - { - return (Task)method.Invoke(controllerPayload.Controller, parameterFactories.Select(p => p(context)).ToArray()); - } - catch (Exception e) - { - AddExceptionData(e); - throw; - } - }; - } - - - private MessageHandlerFunc WrapObjectMethod(MethodBase method, IEnumerable parameterFactories) + private MessageHandlerFunc WrapValueTaskMethod(MethodBase method, IEnumerable parameterFactories) { return context => { var controllerPayload = context.Get(); try { - return Task.FromResult(method.Invoke(controllerPayload.Controller, parameterFactories.Select(p => p(context)).ToArray())); + return (ValueTask)method.Invoke(controllerPayload.Controller, parameterFactories.Select(p => p(context)).ToArray()); } catch (Exception e) { diff --git a/Tapeti/Default/PublishResultBinding.cs b/Tapeti/Default/PublishResultBinding.cs index dd0bee9..80a2f90 100644 --- a/Tapeti/Default/PublishResultBinding.cs +++ b/Tapeti/Default/PublishResultBinding.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Diagnostics; using System.Reflection; using System.Threading.Tasks; @@ -23,7 +23,7 @@ namespace Tapeti.Default return; - var hasClassResult = context.Result.Info.ParameterType.IsTypeOrTaskOf(t => t.IsClass, out var isTaskOf, out var actualType); + var hasClassResult = context.Result.Info.ParameterType.IsTypeOrTaskOf(t => t.IsClass, out var taskType, out var actualType); var request = context.MessageClass?.GetCustomAttribute(); var expectedClassResult = request?.Response; @@ -32,35 +32,55 @@ namespace Tapeti.Default // Tapeti 1.2: if you just want to publish another message as a result of the incoming message, explicitly call IPublisher.Publish. // ReSharper disable once ConvertIfStatementToSwitchStatement if (!hasClassResult && expectedClassResult != null || hasClassResult && expectedClassResult != actualType) - throw new ArgumentException($"Message handler must return type {expectedClassResult?.FullName ?? "void"} in controller {context.Method.DeclaringType?.FullName}, method {context.Method.Name}, found: {actualType?.FullName ?? "void"}"); + throw new ArgumentException($"Message handler for non-request message type {context.MessageClass?.FullName} must return type {expectedClassResult?.FullName ?? "void"} in controller {context.Method.DeclaringType?.FullName}, method {context.Method.Name}, found: {actualType?.FullName ?? "void"}"); if (!hasClassResult) return; - if (isTaskOf) + switch (taskType) { - var handler = GetType().GetMethod("PublishGenericTaskResult", BindingFlags.NonPublic | BindingFlags.Static)?.MakeGenericMethod(actualType); - Debug.Assert(handler != null, nameof(handler) + " != null"); + case TaskType.None: + context.Result.SetHandler((messageContext, value) => Reply(value, messageContext)); + break; - context.Result.SetHandler(async (messageContext, value) => { await (Task) handler.Invoke(null, new[] {messageContext, value }); }); + case TaskType.Task: + var handler = GetType().GetMethod(nameof(PublishGenericTaskResult), BindingFlags.NonPublic | BindingFlags.Static)?.MakeGenericMethod(actualType); + Debug.Assert(handler != null, nameof(handler) + " != null"); + + context.Result.SetHandler((messageContext, value) => (ValueTask)handler.Invoke(null, new[] { messageContext, value })); + break; + + case TaskType.ValueTask: + var valueTaskHandler = GetType().GetMethod(nameof(PublishGenericValueTaskResult), BindingFlags.NonPublic | BindingFlags.Static)?.MakeGenericMethod(actualType); + Debug.Assert(valueTaskHandler != null, nameof(handler) + " != null"); + + context.Result.SetHandler((messageContext, value) => (ValueTask)valueTaskHandler.Invoke(null, new[] { messageContext, value })); + break; + + default: + throw new ArgumentOutOfRangeException(); } - else - context.Result.SetHandler((messageContext, value) => Reply(value, messageContext)); } - // ReSharper disable once UnusedMember.Local - used implicitly above - private static async Task PublishGenericTaskResult(IMessageContext messageContext, object value) where T : class + private static async ValueTask PublishGenericTaskResult(IMessageContext messageContext, object value) where T : class { var message = await (Task)value; await Reply(message, messageContext); } - private static Task Reply(object message, IMessageContext messageContext) + private static async ValueTask PublishGenericValueTaskResult(IMessageContext messageContext, object value) where T : class + { + var message = await (ValueTask)value; + await Reply(message, messageContext); + } + + + private static async ValueTask Reply(object message, IMessageContext messageContext) { if (message == null) throw new ArgumentException("Return value of a request message handler must not be null"); @@ -71,9 +91,10 @@ namespace Tapeti.Default CorrelationId = messageContext.Properties.CorrelationId }; - return !string.IsNullOrEmpty(messageContext.Properties.ReplyTo) - ? publisher.PublishDirect(message, messageContext.Properties.ReplyTo, properties, messageContext.Properties.Persistent.GetValueOrDefault(true)) - : publisher.Publish(message, properties, false); + if (!string.IsNullOrEmpty(messageContext.Properties.ReplyTo)) + await publisher.PublishDirect(message, messageContext.Properties.ReplyTo, properties, messageContext.Properties.Persistent.GetValueOrDefault(true)); + else + await publisher.Publish(message, properties, false); } } } diff --git a/Tapeti/Helpers/MiddlewareHelper.cs b/Tapeti/Helpers/MiddlewareHelper.cs index ba1158e..0bdf9a0 100644 --- a/Tapeti/Helpers/MiddlewareHelper.cs +++ b/Tapeti/Helpers/MiddlewareHelper.cs @@ -45,7 +45,7 @@ namespace Tapeti.Helpers /// Receives the middleware which should be called and a reference to the action which will call the next. Pass this on to the middleware. /// The action to execute when the innermost middleware calls next. /// - public static async Task GoAsync(IReadOnlyList middleware, Func, Task> handle, Func lastHandler) + public static async ValueTask GoAsync(IReadOnlyList middleware, Func, ValueTask> handle, Func lastHandler) { var handlerIndex = middleware?.Count - 1 ?? -1; if (middleware == null || handlerIndex == -1) @@ -54,7 +54,7 @@ namespace Tapeti.Helpers return; } - async Task HandleNext() + async ValueTask HandleNext() { handlerIndex--; if (handlerIndex >= 0) diff --git a/Tapeti/Helpers/TaskTypeHelper.cs b/Tapeti/Helpers/TaskTypeHelper.cs index 44e0c99..49446b9 100644 --- a/Tapeti/Helpers/TaskTypeHelper.cs +++ b/Tapeti/Helpers/TaskTypeHelper.cs @@ -3,30 +3,59 @@ using System.Threading.Tasks; namespace Tapeti.Helpers { + /// + /// Determines if a type is a Task, ValueTask or other type. + /// + public enum TaskType + { + /// + /// Type is not a Task or ValueTask. + /// + None, + + /// + /// Type is a Task or Task<T> + /// + Task, + + /// + /// Type is a ValueTask or ValueTask<T> + /// + ValueTask + } + + /// /// Helper methods for working with synchronous and asynchronous versions of methods. /// public static class TaskTypeHelper { /// - /// Determines if the given type matches the predicate, taking Task types into account. + /// Determines if the given type matches the predicate, taking Task and ValueTask types into account. /// /// /// - /// + /// /// - public static bool IsTypeOrTaskOf(this Type type, Func predicate, out bool isTaskOf, out Type actualType) + public static bool IsTypeOrTaskOf(this Type type, Func predicate, out TaskType taskType, out Type actualType) { if (type == typeof(Task)) { - isTaskOf = false; + taskType = TaskType.Task; + actualType = type; + return false; + } + + if (type == typeof(ValueTask)) + { + taskType = TaskType.ValueTask; actualType = type; return false; } if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(Task<>)) { - isTaskOf = true; + taskType = TaskType.Task; var genericArguments = type.GetGenericArguments(); if (genericArguments.Length == 1 && predicate(genericArguments[0])) @@ -36,7 +65,19 @@ namespace Tapeti.Helpers } } - isTaskOf = false; + if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(ValueTask<>)) + { + taskType = TaskType.ValueTask; + + var genericArguments = type.GetGenericArguments(); + if (genericArguments.Length == 1 && predicate(genericArguments[0])) + { + actualType = genericArguments[0]; + return true; + } + } + + taskType = TaskType.None; actualType = type; return predicate(type); } @@ -47,10 +88,10 @@ namespace Tapeti.Helpers /// /// /// - /// - public static bool IsTypeOrTaskOf(this Type type, Func predicate, out bool isTaskOf) + /// + public static bool IsTypeOrTaskOf(this Type type, Func predicate, out TaskType taskType) { - return IsTypeOrTaskOf(type, predicate, out isTaskOf, out _); + return IsTypeOrTaskOf(type, predicate, out taskType, out _); } @@ -59,10 +100,10 @@ namespace Tapeti.Helpers /// /// /// - /// - public static bool IsTypeOrTaskOf(this Type type, Type compareTo, out bool isTaskOf) + /// + public static bool IsTypeOrTaskOf(this Type type, Type compareTo, out TaskType taskType) { - return IsTypeOrTaskOf(type, t => t == compareTo, out isTaskOf); + return IsTypeOrTaskOf(type, t => t == compareTo, out taskType); } } } diff --git a/Tapeti/Tapeti.csproj b/Tapeti/Tapeti.csproj index 5213bf1..b58e8d9 100644 --- a/Tapeti/Tapeti.csproj +++ b/Tapeti/Tapeti.csproj @@ -30,6 +30,7 @@ + diff --git a/Tapeti/TapetiConfigControllers.cs b/Tapeti/TapetiConfigControllers.cs index f2890de..f4e6b32 100644 --- a/Tapeti/TapetiConfigControllers.cs +++ b/Tapeti/TapetiConfigControllers.cs @@ -43,7 +43,7 @@ namespace Tapeti var controllerIsObsolete = controller.GetCustomAttribute() != null; - foreach (var method in controller.GetMembers(BindingFlags.Public | BindingFlags.Instance) + foreach (var method in controller.GetMembers(BindingFlags.Public | BindingFlags.Instance | BindingFlags.Static) .Where(m => m.MemberType == MemberTypes.Method && m.DeclaringType != typeof(object) && (m as MethodInfo)?.IsSpecialName == false) .Select(m => (MethodInfo)m)) {