WorkflowOptions support for configuring max number of concurrent tasks.

Author: Eddie Wood

src/WorkflowCore/Models/WorkflowOptions.cs

@@ -8,6 +8,8 @@ namespace WorkflowCore.Models
 {
     public class WorkflowOptions
     {
+        internal static readonly int MinimumNumberOfConcurrentItems = 2;
+
         internal Func<IServiceProvider, IPersistenceProvider> PersistanceFactory;
         internal Func<IServiceProvider, IQueueProvider> QueueFactory;
         internal Func<IServiceProvider, IDistributedLockProvider> LockFactory;
@@ -16,6 +18,7 @@ public class WorkflowOptions
         internal TimeSpan PollInterval;
         internal TimeSpan IdleTime;
         internal TimeSpan ErrorRetryInterval;
+        internal int? MaxConcurrentItems;
 
         public IServiceCollection Services { get; private set; }
 
@@ -67,6 +70,16 @@ public void UseErrorRetryInterval(TimeSpan interval)
         {
             ErrorRetryInterval = interval;
         }
+
+        public void UseMaxConcurrentItems(int? maxConcurrentItems)
+        {
+            if (maxConcurrentItems.HasValue && maxConcurrentItems.Value < MinimumNumberOfConcurrentItems)
+            {
+                throw new ArgumentOutOfRangeException($"If {nameof(maxConcurrentItems)} is specified, it cannot be less than ${MinimumNumberOfConcurrentItems}.");
+            }
+
+            MaxConcurrentItems = maxConcurrentItems;
+        }
     }
 
 }

src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs

@@ -13,7 +13,7 @@ internal class EventConsumer : QueueConsumer, IBackgroundTask
         private readonly IPersistenceProvider _persistenceStore;
         private readonly IDistributedLockProvider _lockProvider;
         private readonly IDateTimeProvider _datetimeProvider;
-        protected override int MaxConcurrentItems => 2;
+        protected override int MaxConcurrentItems => WorkflowOptions.MinimumNumberOfConcurrentItems;
         protected override QueueType Queue => QueueType.Event;
 
         public EventConsumer(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider)

src/WorkflowCore/Services/BackgroundTasks/QueueConsumer.cs

@@ -11,7 +11,7 @@ namespace WorkflowCore.Services.BackgroundTasks
     internal abstract class QueueConsumer : IBackgroundTask
     {
         protected abstract QueueType Queue { get; }
-        protected virtual int MaxConcurrentItems => Math.Max(Environment.ProcessorCount, 2);
+        protected virtual int MaxConcurrentItems => Options.MaxConcurrentItems ?? Math.Max(Environment.ProcessorCount, WorkflowOptions.MinimumNumberOfConcurrentItems);
         protected virtual bool EnableSecondPasses => false;
 
         protected readonly IQueueProvider QueueProvider;