* Move code for handling ack, nack, and return into Channel.PublisherConfirms.cs

Author: lukebakken

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

@@ -29,10 +29,15 @@
 //  Copyright (c) 2007-2024 Broadcom. All Rights Reserved.
 //---------------------------------------------------------------------------
 
+using System.Buffers.Binary;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Diagnostics;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
+using RabbitMQ.Client.Events;
+using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing;
 
 namespace RabbitMQ.Client.Impl
@@ -92,5 +97,74 @@ await ModelSendAsync(in method, k.CancellationToken)
                 }
             }
         }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void HandleAck(ulong deliveryTag, bool multiple)
+        {
+            if (_publisherConfirmationsEnabled && _publisherConfirmationTrackingEnabled && deliveryTag > 0 && !_confirmsTaskCompletionSources.IsEmpty)
+            {
+                if (multiple)
+                {
+                    foreach (KeyValuePair<ulong, TaskCompletionSource<bool>> pair in _confirmsTaskCompletionSources)
+                    {
+                        if (pair.Key <= deliveryTag)
+                        {
+                            pair.Value.SetResult(true);
+                            _confirmsTaskCompletionSources.Remove(pair.Key, out _);
+                        }
+                    }
+                }
+                else
+                {
+                    if (_confirmsTaskCompletionSources.TryRemove(deliveryTag, out TaskCompletionSource<bool>? tcs))
+                    {
+                        tcs.SetResult(true);
+                    }
+                }
+            }
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void HandleNack(ulong deliveryTag, bool multiple, bool isReturn)
+        {
+            if (_publisherConfirmationsEnabled && _publisherConfirmationTrackingEnabled && deliveryTag > 0 && !_confirmsTaskCompletionSources.IsEmpty)
+            {
+                if (multiple)
+                {
+                    foreach (KeyValuePair<ulong, TaskCompletionSource<bool>> pair in _confirmsTaskCompletionSources)
+                    {
+                        if (pair.Key <= deliveryTag)
+                        {
+                            pair.Value.SetException(new PublishException(pair.Key, isReturn));
+                            _confirmsTaskCompletionSources.Remove(pair.Key, out _);
+                        }
+                    }
+                }
+                else
+                {
+                    if (_confirmsTaskCompletionSources.Remove(deliveryTag, out TaskCompletionSource<bool>? tcs))
+                    {
+                        tcs.SetException(new PublishException(deliveryTag, isReturn));
+                    }
+                }
+            }
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void HandleReturn(BasicReturnEventArgs basicReturnEvent)
+        {
+            if (_publisherConfirmationsEnabled)
+            {
+                ulong publishSequenceNumber = 0;
+                IReadOnlyBasicProperties props = basicReturnEvent.BasicProperties;
+                object? maybeSeqNum = props.Headers?[Constants.PublishSequenceNumberHeader];
+                if (maybeSeqNum != null)
+                {
+                    publishSequenceNumber = BinaryPrimitives.ReadUInt64BigEndian((byte[])maybeSeqNum);
+                }
+
+                HandleNack(publishSequenceNumber, multiple: false, isReturn: true);
+            }
+        }
     }
 }

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -30,7 +30,6 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Buffers.Binary;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
@@ -602,8 +601,7 @@ await _basicAcksAsyncWrapper.InvokeAsync(this, args)
                     .ConfigureAwait(false);
             }
 
-            await HandleAck(ack._deliveryTag, ack._multiple, cancellationToken)
-                .ConfigureAwait(false);
+            HandleAck(ack._deliveryTag, ack._multiple);
 
             return true;
         }
@@ -620,8 +618,7 @@ await _basicNacksAsyncWrapper.InvokeAsync(this, args)
                     .ConfigureAwait(false);
             }
 
-            await HandleNack(nack._deliveryTag, nack._multiple, false, cancellationToken)
-                .ConfigureAwait(false);
+            HandleNack(nack._deliveryTag, nack._multiple, false);
 
             return true;
         }
@@ -640,19 +637,7 @@ await _basicReturnAsyncWrapper.InvokeAsync(this, e)
                     .ConfigureAwait(false);
             }
 
-            if (_publisherConfirmationsEnabled)
-            {
-                ulong publishSequenceNumber = 0;
-                IReadOnlyBasicProperties props = e.BasicProperties;
-                object? maybeSeqNum = props.Headers?[Constants.PublishSequenceNumberHeader];
-                if (maybeSeqNum != null)
-                {
-                    publishSequenceNumber = BinaryPrimitives.ReadUInt64BigEndian((byte[])maybeSeqNum);
-                }
-
-                await HandleNack(publishSequenceNumber, multiple: false, isReturn: true, cancellationToken)
-                    .ConfigureAwait(false);
-            }
+            HandleReturn(e);
 
             return true;
         }
@@ -1716,61 +1701,6 @@ await ModelSendAsync(in method, k.CancellationToken)
             }
         }
 
-        private Task HandleAck(ulong deliveryTag, bool multiple, CancellationToken cancellationToken = default)
-        {
-            if (_publisherConfirmationsEnabled && _publisherConfirmationTrackingEnabled && deliveryTag > 0 && !_confirmsTaskCompletionSources.IsEmpty)
-            {
-                if (multiple)
-                {
-                    foreach (KeyValuePair<ulong, TaskCompletionSource<bool>> pair in _confirmsTaskCompletionSources)
-                    {
-                        if (pair.Key <= deliveryTag)
-                        {
-                            pair.Value.SetResult(true);
-                            _confirmsTaskCompletionSources.Remove(pair.Key, out _);
-                        }
-                    }
-                }
-                else
-                {
-                    if (_confirmsTaskCompletionSources.TryRemove(deliveryTag, out TaskCompletionSource<bool>? tcs))
-                    {
-                        tcs.SetResult(true);
-                    }
-                }
-            }
-
-            return Task.CompletedTask;
-        }
-
-        private Task HandleNack(ulong deliveryTag, bool multiple, bool isReturn,
-            CancellationToken cancellationToken = default)
-        {
-            if (_publisherConfirmationsEnabled && _publisherConfirmationTrackingEnabled && deliveryTag > 0 && !_confirmsTaskCompletionSources.IsEmpty)
-            {
-                if (multiple)
-                {
-                    foreach (KeyValuePair<ulong, TaskCompletionSource<bool>> pair in _confirmsTaskCompletionSources)
-                    {
-                        if (pair.Key <= deliveryTag)
-                        {
-                            pair.Value.SetException(new PublishException(pair.Key, isReturn));
-                            _confirmsTaskCompletionSources.Remove(pair.Key, out _);
-                        }
-                    }
-                }
-                else
-                {
-                    if (_confirmsTaskCompletionSources.Remove(deliveryTag, out TaskCompletionSource<bool>? tcs))
-                    {
-                        tcs.SetException(new PublishException(deliveryTag, isReturn));
-                    }
-                }
-            }
-
-            return Task.CompletedTask;
-        }
-
         private BasicProperties? PopulateBasicPropertiesHeaders<TProperties>(TProperties basicProperties,
             Activity? sendActivity, ulong publishSequenceNumber)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader