Feature/cdms 188 configure concurrency

Btms.Business.Tests/Commands/SyncClearanceRequestsCommandTests.cs

@@ -51,7 +51,7 @@ public async Task WhenClearanceRequestBlobsExist_ThenTheyShouldBePlacedOnInterna
             await handler.Handle(command, CancellationToken.None);
 
             // ASSERT
-            await bus.Received(1).Publish(Arg.Any<AlvsClearanceRequest>(), "ALVS",
+            await bus.Received(1).Publish(Arg.Any<AlvsClearanceRequest>(), "CLEARANCEREQUESTS",
                 Arg.Any<IDictionary<string, object>>(), Arg.Any<CancellationToken>());
         }
     }

Btms.Business/BusinessOptions.cs

@@ -1,12 +1,49 @@
 using System.ComponentModel.DataAnnotations;
 using Btms.Azure;
+using Btms.Business.Commands;
 
 namespace Btms.Business;
 
 public class BusinessOptions
 {
     public const string SectionName = nameof(BusinessOptions);
 
-    [Required] public string DmpBlobRootFolder { get; set; } = "RAW";
+    private readonly int defaultDegreeOfParallelism = Math.Max(Environment.ProcessorCount / 4, 1);
 
+    [Required] public string DmpBlobRootFolder { get; set; } = "RAW";
+
+    public Dictionary<string, Dictionary<Feature, int>> ConcurrencyConfiguration { get; set; }
+
+    public enum Feature
+    {
+        BlobPaths,
+        BlobItems
+    }
+    public BusinessOptions()
+    {
+        ConcurrencyConfiguration = new Dictionary<string, Dictionary<Feature, int>>
+        {
+            {
+                nameof(SyncNotificationsCommand), new Dictionary<Feature, int>()
+                {
+                    { Feature.BlobPaths, defaultDegreeOfParallelism }, { Feature.BlobItems, defaultDegreeOfParallelism }
+                }
+            },
+            {
+                nameof(SyncClearanceRequestsCommand), new Dictionary<Feature, int>()
+                {
+                    { Feature.BlobPaths, defaultDegreeOfParallelism }, { Feature.BlobItems, defaultDegreeOfParallelism }
+                }
+            }
+        };
+    }
+
+    public int GetConcurrency<T>(Feature feature)
+    {
+        if (ConcurrencyConfiguration.TryGetValue(typeof(T).Name, out var degreeOfParallelismDictionary))
+        {
+            return degreeOfParallelismDictionary.GetValueOrDefault(feature, defaultDegreeOfParallelism);
+        }
+        return defaultDegreeOfParallelism;
+    }
 }

Btms.Business/Commands/SyncClearanceRequestsCommand.cs

@@ -20,14 +20,14 @@ internal class Handler(
         IOptions<BusinessOptions> businessOptions,
         ISyncJobStore syncJobStore)
         : SyncCommand.Handler<SyncClearanceRequestsCommand>(syncMetrics, bus, logger, sensitiveDataSerializer,
-            blobService, syncJobStore)
+            blobService, businessOptions, syncJobStore)
     {
         public override async Task Handle(SyncClearanceRequestsCommand request, CancellationToken cancellationToken)
         {
             var rootFolder = string.IsNullOrEmpty(request.RootFolder)
-                ? businessOptions.Value.DmpBlobRootFolder
+                ? Options.DmpBlobRootFolder
                 : request.RootFolder;
-            await SyncBlobPaths<AlvsClearanceRequest>(request.SyncPeriod, "ALVS", request.JobId, cancellationToken,$"{rootFolder}/ALVS");
+            await SyncBlobPaths<AlvsClearanceRequest>(request.SyncPeriod, "CLEARANCEREQUESTS", request.JobId, cancellationToken,$"{rootFolder}/ALVS");
         }
     }
 

Btms.Business/Commands/SyncDecisionsCommand.cs

@@ -19,12 +19,12 @@ internal class Handler(
         IBlobService blobService,
         IOptions<BusinessOptions> businessOptions,
         ISyncJobStore syncJobStore)
-        : SyncCommand.Handler<SyncDecisionsCommand>(syncMetrics, bus, logger, sensitiveDataSerializer, blobService, syncJobStore)
+        : SyncCommand.Handler<SyncDecisionsCommand>(syncMetrics, bus, logger, sensitiveDataSerializer, blobService, businessOptions, syncJobStore)
     {
         public override async Task Handle(SyncDecisionsCommand request, CancellationToken cancellationToken)
         {
             var rootFolder = string.IsNullOrEmpty(request.RootFolder)
-                ? businessOptions.Value.DmpBlobRootFolder
+                ? Options.DmpBlobRootFolder
                 : request.RootFolder;
             await SyncBlobPaths<AlvsClearanceRequest>(request.SyncPeriod, "DECISIONS", request.JobId,
                 cancellationToken,$"{rootFolder}/DECISIONS");

Btms.Business/Commands/SyncGmrsCommand.cs

@@ -19,12 +19,12 @@ internal class Handler(
         IBlobService blobService,
         IOptions<BusinessOptions> businessOptions,
         ISyncJobStore syncJobStore)
-        : SyncCommand.Handler<SyncGmrsCommand>(syncMetrics, bus, logger, sensitiveDataSerializer, blobService, syncJobStore)
+        : SyncCommand.Handler<SyncGmrsCommand>(syncMetrics, bus, logger, sensitiveDataSerializer, blobService, businessOptions, syncJobStore)
     {
         public override async Task Handle(SyncGmrsCommand request, CancellationToken cancellationToken)
         {
             var rootFolder = string.IsNullOrEmpty(request.RootFolder)
-                ? businessOptions.Value.DmpBlobRootFolder
+                ? Options.DmpBlobRootFolder
                 : request.RootFolder;
             await SyncBlobPaths<SearchGmrsForDeclarationIdsResponse>(request.SyncPeriod, "GMR", request.JobId,
                 cancellationToken, $"{rootFolder}/GVMSAPIRESPONSE");

Btms.Business/Commands/SyncHandler.cs

@@ -6,6 +6,7 @@
 using System.Diagnostics;
 using System.Text.Json.Serialization;
 using Btms.Metrics;
+using Microsoft.Extensions.Options;
 using IRequest = MediatR.IRequest;
 
 namespace Btms.Business.Commands;
@@ -51,12 +52,14 @@ internal abstract class Handler<T>(
         ILogger<T> logger,
         ISensitiveDataSerializer sensitiveDataSerializer,
         IBlobService blobService,
+        IOptions<BusinessOptions> options,
         ISyncJobStore syncJobStore)
         : MediatR.IRequestHandler<T>
         where T : IRequest
     {
-        private readonly int maxDegreeOfParallelism = Math.Max(Environment.ProcessorCount / 4, 1);
-
+        // private readonly int defaultDegreeOfParallelism = Math.Max(Environment.ProcessorCount / 4, 1);
+        protected readonly BusinessOptions Options = options.Value; 
+
         public const string ActivityName = "Btms.ProcessBlob";
 
         public abstract Task Handle(T request, CancellationToken cancellationToken);
@@ -65,20 +68,21 @@ protected async Task SyncBlobPaths<TRequest>(SyncPeriod period, string topic, Gu
         {
             var job = syncJobStore.GetJob(jobId);
             job?.Start();
+            var degreeOfParallelism = options.Value.GetConcurrency<T>(BusinessOptions.Feature.BlobPaths);
             using (logger.BeginScope(new List<KeyValuePair<string, object>>
                    {
                        new("JobId", job?.JobId!),
                        new("SyncPeriod", period.ToString()),
-                       new("Parallelism", maxDegreeOfParallelism),
+                       new("Parallelism", degreeOfParallelism),
                        new("ProcessorCount", Environment.ProcessorCount),
                        new("Command", typeof(T).Name),
                    }))
             {
-                logger.SyncStarted(job?.JobId.ToString()!, period.ToString(), maxDegreeOfParallelism, Environment.ProcessorCount, typeof(T).Name);
+                logger.SyncStarted(job?.JobId.ToString()!, period.ToString(), degreeOfParallelism, Environment.ProcessorCount, typeof(T).Name);
                 try
                 {
                     await Parallel.ForEachAsync(paths,
-                        new ParallelOptions() { MaxDegreeOfParallelism = maxDegreeOfParallelism },
+                        new ParallelOptions() { MaxDegreeOfParallelism = degreeOfParallelism },
                         async (path, token) =>
                         {
                             using (logger.BeginScope(new List<KeyValuePair<string, object>> { new("SyncPath", path), }))
@@ -104,8 +108,9 @@ protected async Task SyncBlobPath<TRequest>(string path, SyncPeriod period, stri
             CancellationToken cancellationToken)
         {
             var result = blobService.GetResourcesAsync($"{path}{period.GetPeriodPath()}", cancellationToken);
+            var degreeOfParallelism = options.Value.GetConcurrency<T>(BusinessOptions.Feature.BlobItems);
 
-            await Parallel.ForEachAsync(result, new ParallelOptions() { CancellationToken = cancellationToken, MaxDegreeOfParallelism = maxDegreeOfParallelism }, async (item, token) =>
+            await Parallel.ForEachAsync(result, new ParallelOptions() { CancellationToken = cancellationToken, MaxDegreeOfParallelism = degreeOfParallelism }, async (item, token) =>
             {
                 await SyncBlob<TRequest>(path, topic, item, job, cancellationToken);
             });
@@ -115,8 +120,10 @@ protected async Task SyncBlobPath<TRequest>(string path, SyncPeriod period, stri
         protected async Task SyncBlobs<TRequest>(SyncPeriod period, string topic, Guid jobId, CancellationToken cancellationToken, params string[] paths)
         {
             var job = syncJobStore.GetJob(jobId);
+            var degreeOfParallelism = options.Value.GetConcurrency<T>(BusinessOptions.Feature.BlobItems);
+
             job?.Start();
-            logger.LogInformation("SyncNotifications period: {Period}, maxDegreeOfParallelism={MaxDegreeOfParallelism}, Environment.ProcessorCount={ProcessorCount}", period.ToString(), maxDegreeOfParallelism, Environment.ProcessorCount);
+            logger.LogInformation("SyncNotifications period: {Period}, maxDegreeOfParallelism={degreeOfParallelism}, Environment.ProcessorCount={ProcessorCount}", period.ToString(), degreeOfParallelism, Environment.ProcessorCount);
             try
             {
                 foreach (var path in paths)

Btms.Business/Commands/SyncNotificationsCommand.cs

@@ -27,12 +27,12 @@ internal class Handler(
             IOptions<BusinessOptions> businessOptions,
             ISyncJobStore syncJobStore)
             : SyncCommand.Handler<SyncNotificationsCommand>(syncMetrics, bus, logger, sensitiveDataSerializer,
-                blobService, syncJobStore)
+                blobService, businessOptions, syncJobStore)
         {
             public override async Task Handle(SyncNotificationsCommand request, CancellationToken cancellationToken)
             {
                 var rootFolder = string.IsNullOrEmpty(request.RootFolder)
-                    ? businessOptions.Value.DmpBlobRootFolder
+                    ? Options.DmpBlobRootFolder
                     : request.RootFolder;
 
                 if (request.BlobFiles.Any())

Btms.Consumers/ConsumerOptions.cs

@@ -0,0 +1,15 @@
+using System.ComponentModel.DataAnnotations;
+using Btms.Azure;
+
+namespace Btms.Consumers;
+
+public class ConsumerOptions
+{
+    public const string SectionName = nameof(ConsumerOptions);
+
+    public int InMemoryNotifications { get; set; } = 2;
+    public int InMemoryGmrs { get; set; } = 2;
+    public int InMemoryClearanceRequests { get; set; } = 2;
+    public int InMemoryDecisions { get; set; } = 2;
+
+}

Btms.Consumers/Extensions/ServiceCollectionExtensions.cs

@@ -1,10 +1,13 @@
+using System.Configuration;
+using Btms.Common.Extensions;
 using Btms.Consumers.Interceptors;
 using Btms.Consumers.MemoryQueue;
 using Btms.Metrics.Extensions;
 using Btms.Types.Gvms;
 using Btms.Types.Ipaffs;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
 using SlimMessageBus.Host;
 using SlimMessageBus.Host.Interceptor;
 using SlimMessageBus.Host.Memory;
@@ -17,6 +20,17 @@ public static class ServiceCollectionExtensions
         public static IServiceCollection AddConsumers(this IServiceCollection services,
             IConfiguration configuration)
         {
+            services.BtmsAddOptions<ConsumerOptions>(configuration, ConsumerOptions.SectionName);
+
+            var consumerOpts = configuration
+                .GetSection(ConsumerOptions.SectionName)
+                .Get<ConsumerOptions>() ?? new ConsumerOptions();
+
+            // services.BtmsAddOptions<ConsumerOptions>(configuration, ConsumerOptions.SectionName);
+            //
+            // var consumerOpts = services.GetRequiredService<IOptions<ConsumerOptions>>();
+
+
             services.AddBtmsMetrics();
             services.AddSingleton<IMemoryQueueStatsMonitor, MemoryQueueStatsMonitor>();
             services.AddSingleton(typeof(IConsumerInterceptor<>), typeof(MetricsInterceptor<>));
@@ -42,24 +56,24 @@ public static IServiceCollection AddConsumers(this IServiceCollection services,
                             .Produce<ImportNotification>(x => x.DefaultTopic("NOTIFICATIONS"))
                             .Consume<ImportNotification>(x =>
                             {
-                                x.Instances(2);
+                                x.Instances(consumerOpts.InMemoryNotifications);
                                 x.Topic("NOTIFICATIONS").WithConsumer<NotificationConsumer>();
                             })
                             .Produce<SearchGmrsForDeclarationIdsResponse>(x => x.DefaultTopic("GMR"))
                             .Consume<SearchGmrsForDeclarationIdsResponse>(x =>
                             {
-                                x.Instances(2);
+                                x.Instances(consumerOpts.InMemoryGmrs);
                                 x.Topic("GMR").WithConsumer<GmrConsumer>();
                             })
                             .Produce<AlvsClearanceRequest>(x => x.DefaultTopic(nameof(AlvsClearanceRequest)))
                             .Consume<AlvsClearanceRequest>(x =>
                             {
-                                x.Instances(2);
-                                x.Topic("ALVS").WithConsumer<AlvsClearanceRequestConsumer>();
+                                x.Instances(consumerOpts.InMemoryClearanceRequests);
+                                x.Topic("CLEARANCEREQUESTS").WithConsumer<AlvsClearanceRequestConsumer>();
                             })
                             .Consume<AlvsClearanceRequest>(x =>
                             {
-                                x.Instances(2);
+                                x.Instances(consumerOpts.InMemoryDecisions);
                                 x.Topic("DECISIONS").WithConsumer<DecisionsConsumer>();
                             });
                     });