Update the processing capacity consistently together with the prefetch count to make sure during throttling the number of prefetched messages is kept aligned with the settings of the user (#1058)

Co-authored-by: danielmarbach <danielmarbach@users.noreply.github.com>

Author: danielmarbach

src/Transport/Receiving/MessagePump.cs

@@ -67,11 +67,9 @@ public Task Initialize(
 
         public async Task StartReceive(CancellationToken cancellationToken = default)
         {
-            int prefetchCount = CalculatePrefetchCount();
-
             var receiveOptions = new ServiceBusProcessorOptions
             {
-                PrefetchCount = prefetchCount,
+                PrefetchCount = CalculatePrefetchCount(limitations.MaxConcurrency),
                 ReceiveMode = TransactionMode == TransportTransactionMode.None
                     ? ServiceBusReceiveMode.ReceiveAndDelete
                     : ServiceBusReceiveMode.PeekLock,
@@ -96,20 +94,18 @@ public async Task StartReceive(CancellationToken cancellationToken = default)
                 {
                     criticalErrorAction("Failed to receive message from Azure Service Bus.", ex,
                         messageProcessingCancellationTokenSource.Token);
-                }, () =>
-                    //We don't have to update the prefetch count since we are failing to receive anyway
-                    processor.UpdateConcurrency(1),
-                () => processor.UpdateConcurrency(limitations.MaxConcurrency));
+                }, () => UpdateProcessingCapacity(1),
+                () => UpdateProcessingCapacity(limitations.MaxConcurrency));
 
             await processor.StartProcessingAsync(cancellationToken)
                 .ConfigureAwait(false);
         }
 
         TransportTransactionMode TransactionMode => transportSettings.TransportTransactionMode;
 
-        int CalculatePrefetchCount()
+        int CalculatePrefetchCount(int maxConcurrency)
         {
-            var prefetchCount = limitations.MaxConcurrency * transportSettings.PrefetchMultiplier;
+            var prefetchCount = maxConcurrency * transportSettings.PrefetchMultiplier;
 
             if (transportSettings.PrefetchCount.HasValue)
             {
@@ -190,14 +186,17 @@ public Task ChangeConcurrency(PushRuntimeSettings newLimitations, CancellationTo
         {
             limitations = newLimitations;
 
-            processor.UpdateConcurrency(limitations.MaxConcurrency);
-
-            int prefetchCount = CalculatePrefetchCount();
-            processor.UpdatePrefetchCount(prefetchCount);
+            UpdateProcessingCapacity(limitations.MaxConcurrency);
 
             return Task.CompletedTask;
         }
 
+        void UpdateProcessingCapacity(int maxConcurrency)
+        {
+            processor.UpdateConcurrency(maxConcurrency);
+            processor.UpdatePrefetchCount(CalculatePrefetchCount(maxConcurrency));
+        }
+
         public async Task StopReceive(CancellationToken cancellationToken = default)
         {
             // Wiring up the stop token to trigger the cancellation token that is being