Added ResetStopwatch functionality to IDiagnosticContext

Tapeti.Flow/Default/FlowBindingMiddleware.cs

@@ -148,7 +148,10 @@ namespace Tapeti.Flow.Default
 
         private static ValueTask HandleParallelResponse(IMessageContext context)
         {
-            if (context.TryGet<FlowMessageContextPayload>(out var flowPayload) && flowPayload.FlowIsConverging)
+            if (!context.TryGet<FlowMessageContextPayload>(out var flowPayload))
+                return default;
+
+            if (flowPayload.FlowIsConverging)
                 return default;
 
             var flowHandler = context.Config.DependencyResolver.Resolve<IFlowHandler>();

Tapeti.Flow/Default/FlowProvider.cs

@@ -72,7 +72,7 @@ namespace Tapeti.Flow.Default
 
 
         internal async Task<MessageProperties> PrepareRequest(FlowContext context, ResponseHandlerInfo responseHandlerInfo,
-            string convergeMethodName = null, bool convergeMethodTaskSync = false)
+            string? convergeMethodName = null, bool convergeMethodTaskSync = false)
         {
             if (!context.HasFlowStateAndLock)
             {
@@ -101,7 +101,7 @@ namespace Tapeti.Flow.Default
 
 
         internal async Task SendRequest(FlowContext context, object message, ResponseHandlerInfo responseHandlerInfo,
-            string convergeMethodName = null, bool convergeMethodTaskSync = false)
+            string? convergeMethodName = null, bool convergeMethodTaskSync = false)
         {
             var properties = await PrepareRequest(context, responseHandlerInfo, convergeMethodName, convergeMethodTaskSync);
             await context.Store(responseHandlerInfo.IsDurableQueue);

Tapeti.Serilog/Default/DiagnosticContext.cs

@@ -1,4 +1,6 @@
 using System.Collections.Generic;
+using System.Diagnostics;
+using System.Threading;
 using Serilog.Core;
 using Serilog.Events;
 
@@ -10,16 +12,20 @@ namespace Tapeti.Serilog.Default
     public class DiagnosticContext : IDiagnosticContext
     {
         private readonly global::Serilog.ILogger logger;
+        private readonly Stopwatch stopwatch;
         private readonly List<LogEventProperty> properties = new();
+        private int resetCount;
 
         
         /// <summary>
         /// Creates a new instance of a DiagnosticContext
         /// </summary>
         /// <param name="logger">The Serilog ILogger which will be enriched</param>
-        public DiagnosticContext(global::Serilog.ILogger logger)
+        /// <param name="stopwatch">The Stopwatch instance that monitors the run time of the message handler</param>
+        public DiagnosticContext(global::Serilog.ILogger logger, Stopwatch stopwatch)
         {
             this.logger = logger;
+            this.stopwatch = stopwatch;
         }
 
 
@@ -31,6 +37,17 @@ namespace Tapeti.Serilog.Default
         }
 
 
+        /// <inheritdoc />
+        public void ResetStopwatch(bool addToContext = true, string propertyNamePrefix = "stopwatchReset")
+        {
+            var newResetCount = Interlocked.Increment(ref resetCount);
+            if (addToContext)
+                Set(propertyNamePrefix + newResetCount, stopwatch.ElapsedMilliseconds);
+
+            stopwatch.Restart();
+        }
+
+
         /// <summary>
         /// Returns a Serilog ILogger which is enriched with the properties set by this DiagnosticContext
         /// </summary>

Tapeti.Serilog/IDiagnosticContext.cs

@@ -5,8 +5,7 @@
     /// MessageHandlerLogging middleware.
     /// </summary>
     /// <remarks>
-    /// This is a one-to-one copy of the IDiagnosticContext in Serilog.Extensions.Hosting which
+    /// Similar to IDiagnosticContext in Serilog.Extensions.Hosting but slightly extended.
-    /// saves a reference to that package while allowing similar usage within Tapeti message handlers.
     /// </remarks>
     public interface IDiagnosticContext
     {
@@ -19,5 +18,14 @@
         /// <param name="destructureObjects">If true, the value will be serialized as structured
         /// data if possible; if false, the object will be recorded as a scalar or simple array.</param>
         void Set(string propertyName, object value, bool destructureObjects = false);
+
+        /// <summary>
+        /// Resets the timer which is used to monitor how long a message handler takes to complete.
+        /// Useful for example when a message handler is throttled by a rate limiter in the message
+        /// handler method and you want to measure only the time taken after it is allowed to start.
+        /// </summary>
+        /// <param name="addToContext">If true, the time taken until this reset is added to this diagnostic context as an incrementally named property for logging purposes. The value will be the time in milliseconds.</param>
+        /// <param name="propertyNamePrefix">The prefix for the property name(s) when addToContext is true. The number of times ResetStopwatch is called will be appended (stopwatchReset1, stopwatchReset2, etc).</param>
+        void ResetStopwatch(bool addToContext = true, string propertyNamePrefix = "stopwatchReset");
     }
 }

Tapeti.Serilog/Middleware/MessageHandlerLoggingMessageMiddleware.cs

@@ -32,11 +32,11 @@ namespace Tapeti.Serilog.Middleware
         public async ValueTask Handle(IMessageContext context, Func<ValueTask> next)
         {
             var logger = context.Config.DependencyResolver.Resolve<global::Serilog.ILogger>();
-            
+
-            var diagnosticContext = new DiagnosticContext(logger);
+            var stopwatch = new Stopwatch();
+            var diagnosticContext = new DiagnosticContext(logger, stopwatch);
             context.Store(new DiagnosticContextPayload(diagnosticContext));
             
-            var stopwatch = new Stopwatch();
             stopwatch.Start();
             
             await next();