Parallel Execution of Webhook Handlers

src/Deveel.Webhooks.Receiver.AspNetCore/Webhooks/ApplicationBuilderExtensions.cs

@@ -28,6 +28,7 @@ public static class ApplicationBuilderExtensions {
 		/// <typeparam name="TWebhook">The type of the webhook to be received</typeparam>
 		/// <param name="app">The application builder instance</param>
 		/// <param name="path">The relative path to listen for webhook posts</param>
+		/// <param name="options">The options for the execution of the handlers</param>
 		/// <remarks>
 		/// <para>
 		/// The middleware will listen only for POST requests to the given path using

src/Deveel.Webhooks.Receiver.AspNetCore/Webhooks/HandlerExecutionMode.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace Deveel.Webhooks {
+	/// <summary>
+	/// Enumerates the possible execution modes for webhook handles
+	/// when received and processed.
+	/// </summary>
+	public enum HandlerExecutionMode {
+		/// <summary>
+		/// The handlers are executed sequentially, one at a time.
+		/// </summary>
+		Sequential,
+
+		/// <summary>
+		/// The handlers are executed in parallel, all at the same time.
+		/// </summary>
+		Parallel
+	}
+}

src/Deveel.Webhooks.Receiver.AspNetCore/Webhooks/LoggerExtensions.cs

@@ -57,5 +57,9 @@ static partial class LoggerExtensions {
 		[LoggerMessage(EventId = -20223, Level = LogLevel.Error,
 						Message = "It was not possible to receive a webhook for an unhandled error")]
 		public static partial void LogUnhandledReceiveError(this ILogger logger, Exception error);
+
+		[LoggerMessage(EventId = -20227, Level = LogLevel.Error,
+									Message = "Unhandled error while executing the handler of type '{HandlerType}' for webhooks of type '{WebhookType}'")]
+		public static partial void LogUnhandledHandlerError(this ILogger logger, Exception error, Type handlerType, Type webhookType);
     }
 }

src/Deveel.Webhooks.Receiver.AspNetCore/Webhooks/WebhookReceiverMiddleware.cs

@@ -41,6 +41,39 @@ public WebhookReceiverMiddleware(
 
 		private int InvalidStatusCode => options.InvalidStatusCode ?? 400;
 
+		private async Task HandleWebhookAsync(TWebhook webhook, CancellationToken cancellationToken) {
+			if (handlers == null)
+				return;
+
+			var mode = options.ExecutionMode ?? HandlerExecutionMode.Parallel;
+
+			switch (mode) {
+				case HandlerExecutionMode.Sequential:
+					foreach (var handler in handlers) {
+						await ExecuteAsync(handler, webhook, cancellationToken);
+					}
+					break;
+				case HandlerExecutionMode.Parallel:
+					var parallelOptions = new ParallelOptions {
+						CancellationToken = cancellationToken,
+						MaxDegreeOfParallelism = options.MaxParallelThreads ?? Environment.ProcessorCount
+					};
+					await Parallel.ForEachAsync(handlers, parallelOptions, async (handler, token) => {
+						await ExecuteAsync(handler, webhook, token);
+					});
+
+					break;
+			}
+		}
+
+		private async Task ExecuteAsync(IWebhookHandler<TWebhook> handler, TWebhook webhook, CancellationToken cancellationToken) {
+			try {
+				await handler.HandleAsync(webhook, cancellationToken);
+			} catch (Exception ex) {
+				logger.LogUnhandledHandlerError(ex, handler.GetType(), typeof(TWebhook));
+			}
+		}
+
 		public async Task InvokeAsync(HttpContext context, RequestDelegate next) {
 			try {
 				logger.TraceWebhookArrived();
@@ -56,11 +89,7 @@ public async Task InvokeAsync(HttpContext context, RequestDelegate next) {
 				}
 
 				if (handlers != null && result.Successful && result.Webhook != null) {
-					foreach (var handler in handlers) {
-						await handler.HandleAsync(result.Webhook, context.RequestAborted);
-
-						logger.TraceWebhookHandled(handler.GetType());
-					}
+					await HandleWebhookAsync(result.Webhook, context.RequestAborted);
 				}
 
 				await next.Invoke(context);

src/Deveel.Webhooks.Receiver.AspNetCore/Webhooks/WebhookReceiverOptions.cs

@@ -47,5 +47,19 @@ public class WebhookReceiverOptions {
 		/// from the sender is invalid (<c>400</c> by default).
 		/// </summary>
 		public int? InvalidStatusCode { get; set; } = 400;
+
+		/// <summary>
+		/// Gets or sets the execution mode for the handlers
+		/// during the processing of a received webhook 
+		/// (default: <see cref="HandlerExecutionMode.Parallel"/>).
+		/// </summary>
+		public HandlerExecutionMode? ExecutionMode { get; set; } = HandlerExecutionMode.Parallel;
+
+		/// <summary>
+		/// Gets or sets the maximum number of threads to use when
+		/// executing the handlers in parallel. By default the number
+		/// of processors in the machine is used.
+		/// </summary>
+		public int? MaxParallelThreads { get; set; }
 	}
 }