From 2817a6105d8119a419eb78b3d03bf375c861d97e Mon Sep 17 00:00:00 2001
From: Jonathan M <jonathan.melo@gsoft.com>
Date: Mon, 17 Jun 2024 10:28:12 -0400
Subject: [PATCH] Add safety net to prevent requesting over 100 events at once
 (#141)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Implement controlled parallelism when handling domain events

* Fix tests

* Update src/Workleap.DomainEventPropagation.Subscription.PullDelivery/EventPullerService.cs

Co-authored-by: Gérald Barré <meziantou@users.noreply.github.com>

* PR Rename suggestion

* Rename derp

* Rework parallelization

* Use ValueTask since most of the time this will return immediately

* Try to fix CI

* Ensure we don't request more than 100 events at once

* Adjust documentation

---------

Co-authored-by: Gérald Barré <meziantou@users.noreply.github.com>
---
 README.md                                                  | 7 +++++++
 .../EventPullerService.cs                                  | 3 ++-
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/README.md b/README.md
index e53aeec..3a6af6f 100644
--- a/README.md
+++ b/README.md
@@ -200,6 +200,7 @@ services.AddPullDeliverySubscription()
       "TopicEndpoint": "<azure_topic_uri>",
       "TopicName": "<namespace_topic_to_listen_to>"
       "SubscriptionName": "<subscription_name_under_specified_topic>",
+      "MaxDegreeOfParallelism": 10,
       "TopicAccessKey": "<secret_value>", // Can be omitted to use Azure Identity (RBAC)
     }
   }
@@ -217,12 +218,14 @@ services.AddPullDeliverySubscription()
       "TopicEndpoint": "<azure_topic_uri>",
       "TopicName": "<namespace_topic_to_listen_to>"
       "SubscriptionName": "<subscription_name_under_specified_topic>",
+      "MaxDegreeOfParallelism": 10,
       "TopicAccessKey": "<secret_value>", // Can be omitted to use Azure Identity (RBAC)
     },
     "TopicSub2": {
       "TopicEndpoint": "<azure_topic_uri>",
       "TopicName": "<namespace_topic_to_listen_to>"
       "SubscriptionName": "<subscription_name_under_specified_topic>",
+      "MaxDegreeOfParallelism": 10,
       "TopicAccessKey": "<secret_value>", // Can be omitted to use Azure Identity (RBAC)
     }
   }
@@ -240,6 +243,9 @@ services.AddPullDeliverySubscription()
     // Namespace topic subscription name
     options.SubscriptionName = "<subscription_name>";
 
+    // Maximum degree of parallelism for processing events
+    options.MaxDegreeOfParallelism = 10;
+
     // Using an access key        
     options.TopicAccessKey = "<secret_value>";
     
@@ -281,6 +287,7 @@ public class ExampleDomainEventHandler : IDomainEventHandler<ExampleDomainEvent>
 * You may only define one domain event handler per domain event you wish to handle. If you would require more, use the single allowed domain event handler as a facade for multiple operations.
 * Domain event handlers must have idempotent behavior (you could execute it multiple times for the same event and the result would always be the same).
 * If your domain event types and handlers are in dedicated assemblies, you can reference the [Workleap.DomainEventPropagation.Abstractions](https://www.nuget.org/packages/Workleap.DomainEventPropagation.Abstractions) packages in order to avoid a dependency on third-parties like Azure and Microsoft extensions.
+* For improved performance, it is possible to run multiple event handlers in parallel by adjusting the `MaxDegreeOfParallelism` option in the subscription configuration. The default value is 1.
 
 ## Building, releasing and versioning
 
diff --git a/src/Workleap.DomainEventPropagation.Subscription.PullDelivery/EventPullerService.cs b/src/Workleap.DomainEventPropagation.Subscription.PullDelivery/EventPullerService.cs
index 90a795e..d4a4b14 100644
--- a/src/Workleap.DomainEventPropagation.Subscription.PullDelivery/EventPullerService.cs
+++ b/src/Workleap.DomainEventPropagation.Subscription.PullDelivery/EventPullerService.cs
@@ -40,6 +40,7 @@ protected override Task ExecuteAsync(CancellationToken stoppingToken)
     private class EventGridSubscriptionEventPuller
     {
         private const int OutputChannelSize = 5000;
+        private const int MaxEventRequestSize = 100;
 
         private readonly IServiceScopeFactory _serviceScopeFactory;
         private readonly ILogger<EventPullerService> _logger;
@@ -108,7 +109,7 @@ private async IAsyncEnumerable<EventGridClientAdapter.EventGridClientAdapter.Eve
                     continue;
                 }
 
-                var bundles = await this._eventGridTopicSubscription.Client.ReceiveCloudEventsAsync(this._eventGridTopicSubscription.TopicName, this._eventGridTopicSubscription.SubscriptionName, availableHandlers, cancellationToken).ConfigureAwait(false);
+                var bundles = await this._eventGridTopicSubscription.Client.ReceiveCloudEventsAsync(this._eventGridTopicSubscription.TopicName, this._eventGridTopicSubscription.SubscriptionName, Math.Min(availableHandlers, MaxEventRequestSize), cancellationToken).ConfigureAwait(false);
 
                 foreach (var bundle in bundles)
                 {