move instance concurrency config from manifest to transport customisation (#4486)

PhilBastian · danielmarbach · web-flow · commit f35ae99107f4 · 2024-10-03T06:52:43.000+08:00
* move instance concurrency config from manifest to transport customisation

* update approval files

* Remove EndpointType because it is not necessary

* Proper cleanup

---------

Co-authored-by: danielmarbach &lt;danielmarbach@users.noreply.github.com&gt;
diff --git a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt
@@ -24,7 +24,7 @@
   "MaxBodySizeToStore": 102400,
   "InstanceName": "Particular.ServiceControl.Audit",
   "TransportConnectionString": null,
-  "MaximumConcurrencyLevel": 32,
+  "MaximumConcurrencyLevel": null,
   "ServiceControlQueueAddress": "Particular.ServiceControl",
   "TimeToRestartAuditIngestionAfterFailure": "00:01:00",
   "EnableFullTextSearchOnBodies": true
diff --git a/src/ServiceControl.Audit.UnitTests/Infrastructure/When_instance_is_setup.cs b/src/ServiceControl.Audit.UnitTests/Infrastructure/When_instance_is_setup.cs
@@ -73,8 +73,7 @@ public void CustomizeMonitoringEndpoint(EndpointConfiguration endpointConfigurat
 
         public void AddTransportForMonitoring(IServiceCollection services, TransportSettings transportSettings) => throw new NotImplementedException();
 
-        public Task ProvisionQueues(TransportSettings transportSettings,
-            IEnumerable<string> additionalQueues)
+        public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues)
         {
             QueuesCreated = new List<string>(additionalQueues)
             {
@@ -84,8 +83,8 @@ public Task ProvisionQueues(TransportSettings transportSettings,
             return Task.CompletedTask;
         }
 
-        public Task<TransportInfrastructure> CreateTransportInfrastructure(string name, TransportSettings transportSettings, OnMessage onMessage = null,
-            OnError onError = null, Func<string, Exception, Task> onCriticalError = null,
+        public Task<TransportInfrastructure> CreateTransportInfrastructure(string name, TransportSettings transportSettings,
+            OnMessage onMessage = null, OnError onError = null, Func<string, Exception, Task> onCriticalError = null,
             TransportTransactionMode preferredTransactionMode = TransportTransactionMode.ReceiveOnly) =>
             throw new NotImplementedException();
         public string ToTransportQualifiedQueueName(string queueName) => queueName;
diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs
@@ -44,7 +44,12 @@ public AuditIngestion(
             batchDurationMeter = metrics.GetMeter("Audit ingestion - batch processing duration", FrequencyInMilliseconds);
             receivedMeter = metrics.GetCounter("Audit ingestion - received");
 
-            channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(settings.MaximumConcurrencyLevel)
+            if (!transportSettings.MaxConcurrency.HasValue)
+            {
+                throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
+            }
+
+            channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(transportSettings.MaxConcurrency.Value)
             {
                 SingleReader = true,
                 SingleWriter = false,
@@ -199,7 +204,7 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati
 
         async Task Loop()
         {
-            var contexts = new List<MessageContext>(settings.MaximumConcurrencyLevel);
+            var contexts = new List<MessageContext>(transportSettings.MaxConcurrency.Value);
 
             while (await channel.Reader.WaitToReadAsync())
             {
diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs
@@ -45,7 +45,7 @@ public Settings(string transportType = null, string persisterType = null, Loggin
                 Port = SettingsReader.Read(SettingsRootNamespace, "Port", 44444);
             };
 
-            MaximumConcurrencyLevel = SettingsReader.Read(SettingsRootNamespace, "MaximumConcurrencyLevel", TransportManifestLibrary.Find(TransportType)?.DefaultAuditMaximumConcurrencyLevel ?? 32);
+            MaximumConcurrencyLevel = SettingsReader.Read<int?>(SettingsRootNamespace, "MaximumConcurrencyLevel");
             ServiceControlQueueAddress = SettingsReader.Read<string>(SettingsRootNamespace, "ServiceControlQueueAddress");
             TimeToRestartAuditIngestionAfterFailure = GetTimeToRestartAuditIngestionAfterFailure();
             EnableFullTextSearchOnBodies = SettingsReader.Read(SettingsRootNamespace, "EnableFullTextSearchOnBodies", true);
@@ -144,7 +144,7 @@ public int MaxBodySizeToStore
         public string InstanceName { get; init; } = DEFAULT_INSTANCE_NAME;
 
         public string TransportConnectionString { get; set; }
-        public int MaximumConcurrencyLevel { get; set; }
+        public int? MaximumConcurrencyLevel { get; set; }
         public string ServiceControlQueueAddress { get; set; }
 
         public TimeSpan TimeToRestartAuditIngestionAfterFailure { get; set; }
diff --git a/src/ServiceControl.Monitoring.UnitTests/ApprovalFiles/SettingsTests.PlatformSampleSettings.approved.txt b/src/ServiceControl.Monitoring.UnitTests/ApprovalFiles/SettingsTests.PlatformSampleSettings.approved.txt
@@ -14,6 +14,6 @@
   "HttpPort": "9999",
   "EndpointUptimeGracePeriod": "00:00:40",
   "RootUrl": "http://localhost:9999/",
-  "MaximumConcurrencyLevel": 32,
+  "MaximumConcurrencyLevel": null,
   "ServiceControlThroughputDataQueue": "ServiceControl.ThroughputData"
 }
diff --git a/src/ServiceControl.Monitoring/HostApplicationBuilderExtensions.cs b/src/ServiceControl.Monitoring/HostApplicationBuilderExtensions.cs
@@ -89,7 +89,11 @@ static void ConfigureEndpoint(EndpointConfiguration config, Func<ICriticalErrorC
 
         config.GetSettings().Set(settings);
         config.SetDiagnosticsPath(settings.LoggingSettings.LogPath);
-        config.LimitMessageProcessingConcurrencyTo(settings.MaximumConcurrencyLevel);
+        if (!transportSettings.MaxConcurrency.HasValue)
+        {
+            throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
+        }
+        config.LimitMessageProcessingConcurrencyTo(transportSettings.MaxConcurrency.Value);
 
         config.UseSerialization<SystemJsonSerializer>();
         config.UsePersistence<NonDurablePersistence>();
diff --git a/src/ServiceControl.Monitoring/Settings.cs b/src/ServiceControl.Monitoring/Settings.cs
@@ -37,7 +37,7 @@ public Settings(LoggingSettings loggingSettings = null, string transportType = n
             }
 
             EndpointUptimeGracePeriod = TimeSpan.Parse(SettingsReader.Read(SettingsRootNamespace, "EndpointUptimeGracePeriod", "00:00:40"));
-            MaximumConcurrencyLevel = SettingsReader.Read(SettingsRootNamespace, "MaximumConcurrencyLevel", TransportManifestLibrary.Find(TransportType)?.DefaultMonitoringMaximumConcurrencyLevel ?? 32);
+            MaximumConcurrencyLevel = SettingsReader.Read<int?>(SettingsRootNamespace, "MaximumConcurrencyLevel");
             ServiceControlThroughputDataQueue = SettingsReader.Read(SettingsRootNamespace, "ServiceControlThroughputDataQueue", "ServiceControl.ThroughputData");
 
             AssemblyLoadContextResolver = static assemblyPath => new PluginAssemblyLoadContext(assemblyPath);
@@ -64,7 +64,7 @@ public Settings(LoggingSettings loggingSettings = null, string transportType = n
 
         public string RootUrl => $"http://{HttpHostName}:{HttpPort}/";
 
-        public int MaximumConcurrencyLevel { get; set; }
+        public int? MaximumConcurrencyLevel { get; set; }
 
         public string ServiceControlThroughputDataQueue { get; set; }
 
diff --git a/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs b/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs
@@ -10,15 +10,27 @@
 
 public class PostgreSqlTransportCustomization : TransportCustomization<PostgreSqlTransport>
 {
-    protected override void CustomizeTransportForPrimaryEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings) =>
+    protected override void CustomizeTransportForPrimaryEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings)
+    {
         transportDefinition.TransportTransactionMode = TransportTransactionMode.SendsAtomicWithReceive;
 
-    protected override void CustomizeTransportForAuditEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings) =>
+        transportSettings.MaxConcurrency ??= 10;
+    }
+
+    protected override void CustomizeTransportForAuditEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings)
+    {
         transportDefinition.TransportTransactionMode = TransportTransactionMode.ReceiveOnly;
 
-    protected override void CustomizeTransportForMonitoringEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings) =>
+        transportSettings.MaxConcurrency ??= 10;
+    }
+
+    protected override void CustomizeTransportForMonitoringEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings)
+    {
         transportDefinition.TransportTransactionMode = TransportTransactionMode.ReceiveOnly;
 
+        transportSettings.MaxConcurrency ??= 10;
+    }
+
     protected override void AddTransportForPrimaryCore(IServiceCollection services,
         TransportSettings transportSettings) =>
         services.AddSingleton<IBrokerThroughputQuery, PostgreSqlQuery>();
diff --git a/src/ServiceControl.Transports.PostgreSql/transport.manifest b/src/ServiceControl.Transports.PostgreSql/transport.manifest
@@ -5,9 +5,6 @@
       "DisplayName": "PostgreSQL",
       "AssemblyName": "ServiceControl.Transports.PostgreSql",
       "TypeName": "ServiceControl.Transports.PostgreSql.PostgreSqlTransportCustomization, ServiceControl.Transports.PostgreSql",
-      "DefaultPrimaryMaximumConcurrencyLevel": 10,
-      "DefaultAuditMaximumConcurrencyLevel": 10,
-      "DefaultMonitoringMaximumConcurrencyLevel": 10,
       "SampleConnectionString": "Server=<ServerName>;Database=nservicebus;Port=5432;User Id=<Username>;Password=<Password>;Queue Schema=myschema;Subscriptions Table=schema.tablename",
       "AvailableInSCMU": true,
       "Help": "Specify optional 'Queue Schema' to override the default schema. Specify optional 'Subscriptions Table' to override the default subscriptions table location."
diff --git a/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs b/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs
@@ -16,15 +16,25 @@ protected override void CustomizeTransportForPrimaryEndpoint(EndpointConfigurati
             transportDefinition.TransportTransactionMode = TransportTransactionMode.SendsAtomicWithReceive;
             var routing = new RoutingSettings(endpointConfiguration.GetSettings());
             routing.EnableMessageDrivenPubSubCompatibilityMode();
+
+            transportSettings.MaxConcurrency ??= 10;
         }
 
         //Do not EnableMessageDrivenPubSubCompatibilityMode for send-only endpoint
-        protected override void CustomizeTransportForAuditEndpoint(EndpointConfiguration endpointConfiguration, SqlServerTransport transportDefinition, TransportSettings transportSettings) =>
+        protected override void CustomizeTransportForAuditEndpoint(EndpointConfiguration endpointConfiguration, SqlServerTransport transportDefinition, TransportSettings transportSettings)
+        {
             transportDefinition.TransportTransactionMode = TransportTransactionMode.ReceiveOnly;
 
-        protected override void CustomizeTransportForMonitoringEndpoint(EndpointConfiguration endpointConfiguration, SqlServerTransport transportDefinition, TransportSettings transportSettings) =>
+            transportSettings.MaxConcurrency ??= 10;
+        }
+
+        protected override void CustomizeTransportForMonitoringEndpoint(EndpointConfiguration endpointConfiguration, SqlServerTransport transportDefinition, TransportSettings transportSettings)
+        {
             transportDefinition.TransportTransactionMode = TransportTransactionMode.ReceiveOnly;
 
+            transportSettings.MaxConcurrency ??= 10;
+        }
+
         protected override void AddTransportForPrimaryCore(IServiceCollection services,
             TransportSettings transportSettings)
         {
diff --git a/src/ServiceControl.Transports.Tests/ApprovalFiles/APIApprovals.ServiceControlTransport.approved.txt b/src/ServiceControl.Transports.Tests/ApprovalFiles/APIApprovals.ServiceControlTransport.approved.txt
@@ -78,9 +78,6 @@ namespace ServiceControl.Transports
         public TransportManifestDefinition() { }
         public string[] Aliases { get; set; }
         public string AssemblyName { get; set; }
-        public int? DefaultAuditMaximumConcurrencyLevel { get; set; }
-        public int? DefaultMonitoringMaximumConcurrencyLevel { get; set; }
-        public int? DefaultPrimaryMaximumConcurrencyLevel { get; set; }
         public string DisplayName { get; set; }
         public string Location { get; set; }
         public string Name { get; set; }
@@ -99,7 +96,7 @@ namespace ServiceControl.Transports
         public string ConnectionString { get; set; }
         public string EndpointName { get; set; }
         public string ErrorQueue { get; set; }
-        public int MaxConcurrency { get; set; }
+        public int? MaxConcurrency { get; set; }
         public bool RunCustomChecks { get; set; }
         public string TransportType { get; set; }
     }
diff --git a/src/ServiceControl.Transports/TransportCustomization.cs b/src/ServiceControl.Transports/TransportCustomization.cs
@@ -76,29 +76,38 @@ public void CustomizePrimaryEndpoint(EndpointConfiguration endpointConfiguration
         {
             ConfigureDefaultEndpointSettings(endpointConfiguration, transportSettings);
             var transport = CreateTransport(transportSettings);
-            endpointConfiguration.UseTransport(transport);
             CustomizeTransportForPrimaryEndpoint(endpointConfiguration, transport, transportSettings);
+
+            transportSettings.MaxConcurrency ??= 10;
+
+            endpointConfiguration.UseTransport(transport);
         }
 
         public void CustomizeAuditEndpoint(EndpointConfiguration endpointConfiguration, TransportSettings transportSettings)
         {
             ConfigureDefaultEndpointSettings(endpointConfiguration, transportSettings);
             var transport = CreateTransport(transportSettings);
-            endpointConfiguration.UseTransport(transport);
             CustomizeTransportForAuditEndpoint(endpointConfiguration, transport, transportSettings);
 
+            transportSettings.MaxConcurrency ??= 32;
+
             endpointConfiguration.SendOnly();
 
             //DisablePublishing API is available only on TransportExtensions for transports that implement IMessageDrivenPubSub so we need to set settings directly
             endpointConfiguration.GetSettings().Set("NServiceBus.PublishSubscribe.EnablePublishing", false);
+
+            endpointConfiguration.UseTransport(transport);
         }
 
         public void CustomizeMonitoringEndpoint(EndpointConfiguration endpointConfiguration, TransportSettings transportSettings)
         {
             ConfigureDefaultEndpointSettings(endpointConfiguration, transportSettings);
             var transport = CreateTransport(transportSettings);
-            endpointConfiguration.UseTransport(transport);
             CustomizeTransportForMonitoringEndpoint(endpointConfiguration, transport, transportSettings);
+
+            transportSettings.MaxConcurrency ??= 32;
+
+            endpointConfiguration.UseTransport(transport);
         }
 
         protected void ConfigureDefaultEndpointSettings(EndpointConfiguration endpointConfiguration, TransportSettings transportSettings)
@@ -116,6 +125,9 @@ protected void ConfigureDefaultEndpointSettings(EndpointConfiguration endpointCo
 
         public virtual async Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues)
         {
+            // For queue provisioning MaxConcurrency is not relevant. Settings it to 1 to avoid null checks
+            transportSettings.MaxConcurrency ??= 1;
+
             var transport = CreateTransport(transportSettings);
 
             var hostSettings = new HostSettings(
@@ -169,8 +181,13 @@ public async Task<TransportInfrastructure> CreateTransportInfrastructure(string
 
             if (createReceiver)
             {
+                if (!transportSettings.MaxConcurrency.HasValue)
+                {
+                    throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
+                }
+
                 var transportInfrastructureReceiver = transportInfrastructure.Receivers[name];
-                await transportInfrastructureReceiver.Initialize(new PushRuntimeSettings(transportSettings.MaxConcurrency), onMessage, onError, CancellationToken.None);
+                await transportInfrastructureReceiver.Initialize(new PushRuntimeSettings(transportSettings.MaxConcurrency.Value), onMessage, onError, CancellationToken.None);
             }
 
             return transportInfrastructure;
diff --git a/src/ServiceControl.Transports/TransportManifest.cs b/src/ServiceControl.Transports/TransportManifest.cs
@@ -28,10 +28,6 @@ public class TransportManifestDefinition
 
         public string[] Aliases { get; set; } = [];
 
-        public int? DefaultPrimaryMaximumConcurrencyLevel { get; set; }
-        public int? DefaultAuditMaximumConcurrencyLevel { get; set; }
-        public int? DefaultMonitoringMaximumConcurrencyLevel { get; set; }
-
         internal bool IsMatch(string transportType) =>
             string.Equals(TypeName, transportType, StringComparison.Ordinal) // Type names are case-sensitive
             || string.Equals(Name, transportType, StringComparison.OrdinalIgnoreCase)
diff --git a/src/ServiceControl.Transports/TransportSettings.cs b/src/ServiceControl.Transports/TransportSettings.cs
@@ -14,7 +14,7 @@ public class TransportSettings : SettingsHolder
 
         public string EndpointName { get; set; }
 
-        public int MaxConcurrency { get; set; }
+        public int? MaxConcurrency { get; set; }
 
         public bool RunCustomChecks { get; set; }
 
diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt
@@ -39,7 +39,7 @@
   "TransportConnectionString": null,
   "ProcessRetryBatchesFrequency": "00:00:30",
   "TimeToRestartErrorIngestionAfterFailure": "00:01:00",
-  "MaximumConcurrencyLevel": 10,
+  "MaximumConcurrencyLevel": null,
   "RetryHistoryDepth": 10,
   "RemoteInstances": [],
   "DisableHealthChecks": false
diff --git a/src/ServiceControl/Infrastructure/NServiceBusFactory.cs b/src/ServiceControl/Infrastructure/NServiceBusFactory.cs
@@ -66,7 +66,11 @@ public static void Configure(Settings.Settings settings, ITransportCustomization
                 }
             });
 
-            configuration.LimitMessageProcessingConcurrencyTo(settings.MaximumConcurrencyLevel);
+            if (!transportSettings.MaxConcurrency.HasValue)
+            {
+                throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
+            }
+            configuration.LimitMessageProcessingConcurrencyTo(transportSettings.MaxConcurrency.Value);
 
             configuration.Conventions().DefiningEventsAs(t => typeof(IEvent).IsAssignableFrom(t) || IsExternalContract(t));
 
diff --git a/src/ServiceControl/Infrastructure/Settings/Settings.cs b/src/ServiceControl/Infrastructure/Settings/Settings.cs
@@ -57,7 +57,8 @@ public Settings(
             }
 
             ProcessRetryBatchesFrequency = TimeSpan.FromSeconds(30);
-            MaximumConcurrencyLevel = SettingsReader.Read(SettingsRootNamespace, "MaximumConcurrencyLevel", TransportManifestLibrary.Find(TransportType)?.DefaultPrimaryMaximumConcurrencyLevel ?? 10);
+            //TODO: is this overridden by SCMU??
+            MaximumConcurrencyLevel = SettingsReader.Read<int?>(SettingsRootNamespace, "MaximumConcurrencyLevel");
             RetryHistoryDepth = SettingsReader.Read(SettingsRootNamespace, "RetryHistoryDepth", 10);
             AllowMessageEditing = SettingsReader.Read<bool>(SettingsRootNamespace, "AllowMessageEditing");
             NotificationsFilter = SettingsReader.Read<string>(SettingsRootNamespace, "NotificationsFilter");
@@ -172,7 +173,7 @@ public TimeSpan HeartbeatGracePeriod
         public string TransportConnectionString { get; set; }
         public TimeSpan ProcessRetryBatchesFrequency { get; set; }
         public TimeSpan TimeToRestartErrorIngestionAfterFailure { get; set; }
-        public int MaximumConcurrencyLevel { get; set; }
+        public int? MaximumConcurrencyLevel { get; set; }
 
         public int RetryHistoryDepth { get; set; }
 
diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs
@@ -44,7 +44,12 @@ public ErrorIngestion(
             batchSizeMeter = metrics.GetMeter("Error ingestion - batch size");
             batchDurationMeter = metrics.GetMeter("Error ingestion - batch processing duration", FrequencyInMilliseconds);
 
-            channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(settings.MaximumConcurrencyLevel)
+            if (!transportSettings.MaxConcurrency.HasValue)
+            {
+                throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
+            }
+
+            channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(transportSettings.MaxConcurrency.Value)
             {
                 SingleReader = true,
                 SingleWriter = false,
@@ -74,7 +79,7 @@ public async Task StopAsync(CancellationToken cancellationToken)
 
         async Task Loop()
         {
-            var contexts = new List<MessageContext>(settings.MaximumConcurrencyLevel);
+            var contexts = new List<MessageContext>(transportSettings.MaxConcurrency.Value);
 
             while (await channel.Reader.WaitToReadAsync())
             {

Original file line number	Diff line number	Diff line change
`@@ -73,8 +73,7 @@ public void CustomizeMonitoringEndpoint(EndpointConfiguration endpointConfigurat`
`73`	`73`
`74`	`74`	`public void AddTransportForMonitoring(IServiceCollection services, TransportSettings transportSettings) => throw new NotImplementedException();`
`75`	`75`
`76`		`- public Task ProvisionQueues(TransportSettings transportSettings,`
`77`		`- IEnumerable<string> additionalQueues)`
	`76`	`+ public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues)`
`78`	`77`	`{`
`79`	`78`	`QueuesCreated = new List<string>(additionalQueues)`
`80`	`79`	`{`
`@@ -84,8 +83,8 @@ public Task ProvisionQueues(TransportSettings transportSettings,`
`84`	`83`	`return Task.CompletedTask;`
`85`	`84`	`}`
`86`	`85`
`87`		`- public Task<TransportInfrastructure> CreateTransportInfrastructure(string name, TransportSettings transportSettings, OnMessage onMessage = null,`
`88`		`- OnError onError = null, Func<string, Exception, Task> onCriticalError = null,`
	`86`	`+ public Task<TransportInfrastructure> CreateTransportInfrastructure(string name, TransportSettings transportSettings,`
	`87`	`+ OnMessage onMessage = null, OnError onError = null, Func<string, Exception, Task> onCriticalError = null,`
`89`	`88`	`TransportTransactionMode preferredTransactionMode = TransportTransactionMode.ReceiveOnly) =>`
`90`	`89`	`throw new NotImplementedException();`
`91`	`90`	`public string ToTransportQualifiedQueueName(string queueName) => queueName;`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,12 @@ public AuditIngestion(`
`44`	`44`	`batchDurationMeter = metrics.GetMeter("Audit ingestion - batch processing duration", FrequencyInMilliseconds);`
`45`	`45`	`receivedMeter = metrics.GetCounter("Audit ingestion - received");`
`46`	`46`
`47`		`- channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(settings.MaximumConcurrencyLevel)`
	`47`	`+ if (!transportSettings.MaxConcurrency.HasValue)`
	`48`	`+ {`
	`49`	`+ throw new ArgumentException("MaxConcurrency is not set in TransportSettings");`
	`50`	`+ }`
	`51`	`+`
	`52`	`+ channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(transportSettings.MaxConcurrency.Value)`
`48`	`53`	`{`
`49`	`54`	`SingleReader = true,`
`50`	`55`	`SingleWriter = false,`
`@@ -199,7 +204,7 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati`
`199`	`204`
`200`	`205`	`async Task Loop()`
`201`	`206`	`{`
`202`		`- var contexts = new List<MessageContext>(settings.MaximumConcurrencyLevel);`
	`207`	`+ var contexts = new List<MessageContext>(transportSettings.MaxConcurrency.Value);`
`203`	`208`
`204`	`209`	`while (await channel.Reader.WaitToReadAsync())`
`205`	`210`	`{`
Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,6 @@`
`14`	`14`	`"HttpPort": "9999",`
`15`	`15`	`"EndpointUptimeGracePeriod": "00:00:40",`
`16`	`16`	`"RootUrl": "http://localhost:9999/",`
`17`		`- "MaximumConcurrencyLevel": 32,`
	`17`	`+ "MaximumConcurrencyLevel": null,`
`18`	`18`	`"ServiceControlThroughputDataQueue": "ServiceControl.ThroughputData"`
`19`	`19`	`}`