Support for transport concurrency level (#55)

* Support for transport concurrency level (PushRuntimeSettings.MaxConcurrency)

* docs

docs/README.md

@@ -130,3 +130,7 @@ NOTE: It's suggested to not discard messages in production.
 ### Transport Transaction
 
 - `TransportTransactionMode` (TransactionScope, SendsAtomicWithReceive, ReceiveOnly, None. The default value depends on the transport of choice) allows defining the [endpoint message processing transaction guarantees](https://docs.particular.net/transports/transactions).
+
+### Concurrency level
+
+- `MessageProcessingConcurrency` (`int`, defaults to not set) sets the [endpoint concurrency level](https://docs.particular.net/nservicebus/operations/tuning) that determines how many messages the endpoint processes in parallel.

src/Mattox.NServiceBus.Tests/LearningEndpointTests.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Configuration;
 using NServiceBus.Configuration.AdvancedExtensibility;
+using NServiceBus.Transport;
 
 namespace Mattox.NServiceBus.Tests;
 
@@ -353,4 +354,41 @@ public void Setting_transport_transaction_mode_sets_desired_value()
 
  Assert.Equal(expected, transportDefinition.TransportTransactionMode);
  }
+
+ [Fact]
+ public void Setting_concurrency_level_to_non_parsable_value_throws()
+ {
+ var config = new ConfigurationBuilder()
+ .AddInMemoryCollection(new Dictionary<string, string>()
+ {
+ { "NServiceBus:EndpointConfiguration:Transport:MessageProcessingConcurrency", "cannot be parsed" }
+ })
+ .Build();
+
+ Assert.Throws<ArgumentException>(() =>
+ {
+ var endpoint = new LearningEndpoint("my-endpoint", config);
+ EndpointConfiguration endpointConfiguration = endpoint;
+ });
+ }
+
+ [Fact]
+ public void Setting_concurrency_level_sets_the_expected_value()
+ {
+ const int expected = 4;
+ var config = new ConfigurationBuilder()
+ .AddInMemoryCollection(new Dictionary<string, string>()
+ {
+ { "NServiceBus:EndpointConfiguration:Transport:MessageProcessingConcurrency", expected.ToString() }
+ })
+ .Build();
+
+ var endpoint = new LearningEndpoint("my-endpoint", config);
+ EndpointConfiguration endpointConfiguration = endpoint;
+
+ var settings = endpointConfiguration.GetSettings();
+ var pushRuntimeSettings = settings.GetOrDefault<PushRuntimeSettings>("NServiceBus.Transport.PushRuntimeSettings");
+
+ Assert.Equal(expected, pushRuntimeSettings.MaxConcurrency);
+ }
 }

src/Mattox.NServiceBus/NServiceBusEndpoint.cs

@@ -72,6 +72,7 @@ protected virtual void FinalizeConfiguration()
 
  ConfigureTransport(transportConfigurationSection);
  ConfigurePurgeOnStartup(endpointConfiguration, transportConfigurationSection);
+ ConfigureMessageProcessingConcurrency(endpointConfiguration, transportConfigurationSection);
  ConfigureAuditing(endpointConfiguration, endpointConfigurationSection);
  ConfigureRecoverability();
  ConfigureSendOnly(endpointConfiguration, endpointConfigurationSection);
@@ -98,6 +99,21 @@ protected virtual void FinalizeConfiguration()
  endpointConfigurationPreview?.Invoke(endpointConfiguration);
  }
 
+ static void ConfigureMessageProcessingConcurrency(EndpointConfiguration endpointConfiguration,
+ IConfigurationSection? transportConfigurationSection)
+ {
+ // TODO docs
+ if (transportConfigurationSection?["MessageProcessingConcurrency"] is { } messageProcessingConcurrencyValue)
+ {
+ if (!int.TryParse(messageProcessingConcurrencyValue, out var messageProcessingConcurrency))
+ {
+ throw new ArgumentException("MessageProcessingConcurrency value cannot be parsed to a valid integer.");
+ }
+
+ endpointConfiguration.LimitMessageProcessingConcurrencyTo(messageProcessingConcurrency);
+ }
+ }
+
  static void ConfigureAddressingOptions(EndpointConfiguration endpointConfiguration, IConfigurationSection? endpointConfigurationSection)
  {
  var localAddressOverride = endpointConfigurationSection?["LocalAddressOverride"];