* Make ConfirmSelectAsync private and assume that semaphore is held.

rabbitmq · Oct 2, 2024 · e65c6fb · e65c6fb
1 parent 88aa14d
commit e65c6fb
Show file tree

Hide file tree

Showing 2 changed files with 57 additions and 75 deletions.
diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs
@@ -181,12 +181,6 @@ await newChannel.BasicQosAsync(0, _prefetchCountGlobal, true, cancellationToken)
                     .ConfigureAwait(false);
             }
 
-            if (_publisherConfirmationsEnabled)
-            {
-                await newChannel.ConfirmSelectAsync(_publisherConfirmationTrackingEnabled, cancellationToken)
-                    .ConfigureAwait(false);
-            }
-
             if (_usesTransactions)
             {
                 await newChannel.TxSelectAsync(cancellationToken)
@@ -350,18 +344,6 @@ public Task BasicQosAsync(uint prefetchSize, ushort prefetchCount, bool global,
             return _innerChannel.BasicQosAsync(prefetchSize, prefetchCount, global, cancellationToken);
         }
 
-        public Task ConfirmSelectAsync(bool publisherConfirmationTrackingEnabled = false, CancellationToken cancellationToken = default)
-        {
-            /*
-             * Note:
-             * No need to pass this on to InnerChannel, as confirms will have already
-             * been enabled
-             */
-            _publisherConfirmationsEnabled = true;
-            _publisherConfirmationTrackingEnabled = publisherConfirmationTrackingEnabled;
-            return Task.CompletedTask;
-        }
-
         public async Task ExchangeBindAsync(string destination, string source, string routingKey,
             IDictionary<string, object?>? arguments, bool noWait,
             CancellationToken cancellationToken)

diff --git a/projects/RabbitMQ.Client/Impl/ChannelBase.cs b/projects/RabbitMQ.Client/Impl/ChannelBase.cs
@@ -386,13 +386,21 @@ await _rpcSemaphore.WaitAsync(k.CancellationToken)
             try
             {
                 enqueued = Enqueue(k);
+                if (enqueued)
+                {
+                    var method = new ChannelOpen();
+                    await ModelSendAsync(in method, k.CancellationToken)
+                        .ConfigureAwait(false);
 
-                var method = new ChannelOpen();
-                await ModelSendAsync(in method, k.CancellationToken)
-                    .ConfigureAwait(false);
+                    bool result = await k;
+                    Debug.Assert(result);
 
-                bool result = await k;
-                Debug.Assert(result);
+                    if (_publisherConfirmationsEnabled)
+                    {
+                        await ConfirmSelectAsync(publisherConfirmationTrackingEnabled, cancellationToken)
+                            .ConfigureAwait(false);
+                    }
+                }
             }
             finally
             {
@@ -403,13 +411,6 @@ await ModelSendAsync(in method, k.CancellationToken)
                 _rpcSemaphore.Release();
             }
 
-            if (_publisherConfirmationsEnabled)
-            {
-                // TODO bring this back within RPC semaphore
-                await ConfirmSelectAsync(publisherConfirmationTrackingEnabled, cancellationToken)
-                    .ConfigureAwait(false);
-            }
-
             return this;
         }
 
@@ -1244,51 +1245,6 @@ await ModelSendAsync(in method, k.CancellationToken)
             }
         }
 
-        // TODO internal
-        // TODO rpc semaphore held
-        public async Task ConfirmSelectAsync(bool publisherConfirmationTrackingEnablefd = false,
-            CancellationToken cancellationToken = default)
-        {
-            _publisherConfirmationsEnabled = true;
-            _publisherConfirmationTrackingEnabled = publisherConfirmationTrackingEnablefd;
-
-            bool enqueued = false;
-            var k = new ConfirmSelectAsyncRpcContinuation(ContinuationTimeout, cancellationToken);
-
-            await _rpcSemaphore.WaitAsync(k.CancellationToken)
-                .ConfigureAwait(false);
-            try
-            {
-                if (_nextPublishSeqNo == 0UL)
-                {
-                    if (_publisherConfirmationTrackingEnabled)
-                    {
-                        _pendingDeliveryTags.Clear();
-                        _confirmsTaskCompletionSources.Clear();
-                    }
-                    _nextPublishSeqNo = 1;
-                }
-
-                enqueued = Enqueue(k);
-
-                var method = new ConfirmSelect(false);
-                await ModelSendAsync(in method, k.CancellationToken)
-                    .ConfigureAwait(false);
-
-                bool result = await k;
-                Debug.Assert(result);
-                return;
-            }
-            finally
-            {
-                if (false == enqueued)
-                {
-                    k.Dispose();
-                }
-                _rpcSemaphore.Release();
-            }
-        }
-
         public async Task ExchangeBindAsync(string destination, string source, string routingKey,
             IDictionary<string, object?>? arguments, bool noWait,
             CancellationToken cancellationToken)
@@ -1773,6 +1729,50 @@ await ModelSendAsync(in method, k.CancellationToken)
             }
         }
 
+        // NOTE: _rpcSemaphore is held
+        private async Task ConfirmSelectAsync(bool publisherConfirmationTrackingEnablefd = false,
+            CancellationToken cancellationToken = default)
+        {
+            _publisherConfirmationsEnabled = true;
+            _publisherConfirmationTrackingEnabled = publisherConfirmationTrackingEnablefd;
+
+            bool enqueued = false;
+            var k = new ConfirmSelectAsyncRpcContinuation(ContinuationTimeout, cancellationToken);
+
+            try
+            {
+                if (_nextPublishSeqNo == 0UL)
+                {
+                    if (_publisherConfirmationTrackingEnabled)
+                    {
+                        _pendingDeliveryTags.Clear();
+                        _confirmsTaskCompletionSources.Clear();
+                    }
+                    _nextPublishSeqNo = 1;
+                }
+
+                enqueued = Enqueue(k);
+                if (enqueued)
+                {
+                    var method = new ConfirmSelect(false);
+                    await ModelSendAsync(in method, k.CancellationToken)
+                        .ConfigureAwait(false);
+
+                    bool result = await k;
+                    Debug.Assert(result);
+                }
+
+                return;
+            }
+            finally
+            {
+                if (false == enqueued)
+                {
+                    k.Dispose();
+                }
+            }
+        }
+
         // NOTE: this method is internal for its use in this test:
         // TestWaitForConfirmsWithTimeoutAsync_MessageNacked_WaitingHasTimedout_ReturnFalse
         private async Task HandleAckNack(ulong deliveryTag, bool multiple, bool isNack, CancellationToken cancellationToken = default)