* Actually enforce outstanding pub conf limit.

Author: lukebakken

projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs

@@ -56,7 +56,10 @@ public async ValueTask BasicPublishAsync<TProperties>(string exchange, string ro
                     await MaybeStartPublisherConfirmationTracking(cancellationToken)
                         .ConfigureAwait(false);
 
-                await EnforceFlowControlAsync(cancellationToken)
+                await MaybeEnforceFlowControlAsync(cancellationToken)
+                    .ConfigureAwait(false);
+
+                await MaybeEnforceOutstandingPublisherConfirmationsAsync(cancellationToken)
                     .ConfigureAwait(false);
 
                 var cmd = new BasicPublish(exchange, routingKey, mandatory, default);
@@ -111,7 +114,10 @@ public async ValueTask BasicPublishAsync<TProperties>(CachedString exchange, Cac
                     await MaybeStartPublisherConfirmationTracking(cancellationToken)
                         .ConfigureAwait(false);
 
-                await EnforceFlowControlAsync(cancellationToken)
+                await MaybeEnforceFlowControlAsync(cancellationToken)
+                    .ConfigureAwait(false);
+
+                await MaybeEnforceOutstandingPublisherConfirmationsAsync(cancellationToken)
                     .ConfigureAwait(false);
 
                 var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);
@@ -225,7 +231,7 @@ void MaybeAddPublishSequenceNumberToHeaders(IDictionary<string, object?> headers
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        private ValueTask EnforceFlowControlAsync(CancellationToken cancellationToken)
+        private ValueTask MaybeEnforceFlowControlAsync(CancellationToken cancellationToken)
         {
             if (_flowControlBlock.IsSet)
             {

projects/RabbitMQ.Client/Impl/Channel.PublisherConfirms.cs

@@ -141,6 +141,7 @@ private async Task MaybeConfirmSelect(CancellationToken cancellationToken)
                         if (_publisherConfirmationTrackingEnabled)
                         {
                             _confirmsTaskCompletionSources.Clear();
+                            MaybeUnblockPublishers();
                         }
                         _nextPublishSeqNo = 1;
                     }
@@ -186,13 +187,15 @@ private void HandleAck(ulong deliveryTag, bool multiple)
                         {
                             pair.Value.SetResult(true);
                             _confirmsTaskCompletionSources.Remove(pair.Key, out _);
+                            MaybeUnblockPublishers();
                         }
                     }
                 }
                 else
                 {
                     if (_confirmsTaskCompletionSources.TryRemove(deliveryTag, out TaskCompletionSource<bool>? tcs))
                     {
+                        MaybeUnblockPublishers();
                         tcs.SetResult(true);
                     }
                 }
@@ -212,13 +215,15 @@ private void HandleNack(ulong deliveryTag, bool multiple, bool isReturn)
                         {
                             pair.Value.SetException(new PublishException(pair.Key, isReturn));
                             _confirmsTaskCompletionSources.Remove(pair.Key, out _);
+                            MaybeUnblockPublishers();
                         }
                     }
                 }
                 else
                 {
                     if (_confirmsTaskCompletionSources.Remove(deliveryTag, out TaskCompletionSource<bool>? tcs))
                     {
+                        MaybeUnblockPublishers();
                         tcs.SetException(new PublishException(deliveryTag, isReturn));
                     }
                 }
@@ -261,6 +266,7 @@ await _confirmSemaphore.WaitAsync(reason.CancellationToken)
                         }
 
                         _confirmsTaskCompletionSources.Clear();
+                        MaybeUnblockPublishers();
                     }
                 }
                 finally
@@ -285,6 +291,7 @@ await _confirmSemaphore.WaitAsync(cancellationToken)
                 {
                     publisherConfirmationTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
                     _confirmsTaskCompletionSources[publishSequenceNumber] = publisherConfirmationTcs;
+                    MaybeBlockPublishers();
                 }
 
                 _nextPublishSeqNo++;
@@ -311,6 +318,7 @@ private bool MaybeHandleExceptionWithEnabledPublisherConfirmations(PublisherConf
                 if (_publisherConfirmationTrackingEnabled)
                 {
                     _confirmsTaskCompletionSources.TryRemove(publisherConfirmationInfo.PublishSequenceNumber, out _);
+                    MaybeUnblockPublishers();
                 }
 
                 exceptionWasHandled = publisherConfirmationInfo.MaybeHandleException(ex);
@@ -334,5 +342,49 @@ await publisherConfirmationInfo.MaybeWaitForConfirmationAsync(cancellationToken)
                 }
             }
         }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private ValueTask MaybeEnforceOutstandingPublisherConfirmationsAsync(CancellationToken cancellationToken)
+        {
+            if (_publisherConfirmationTrackingEnabled)
+            {
+                if (_maxOutstandingPublisherConfirmsReached.IsSet)
+                {
+                    return default;
+                }
+                else
+                {
+                    return _maxOutstandingPublisherConfirmsReached.WaitAsync(cancellationToken);
+                }
+            }
+
+            return default;
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void MaybeBlockPublishers()
+        {
+            if (_publisherConfirmationTrackingEnabled)
+            {
+                if (_maxOutstandingPublisherConfirmations is not null
+                    && _confirmsTaskCompletionSources.Count >= _maxOutstandingPublisherConfirmations)
+                {
+                    _maxOutstandingPublisherConfirmsReached.Reset();
+                }
+            }
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void MaybeUnblockPublishers()
+        {
+            if (_publisherConfirmationTrackingEnabled)
+            {
+                if (_maxOutstandingPublisherConfirmations is not null
+                    && _confirmsTaskCompletionSources.Count < _maxOutstandingPublisherConfirmations)
+                {
+                    _maxOutstandingPublisherConfirmsReached.Set();
+                }
+            }
+        }
     }
 }