* Move more code to Channel.BasicPublish.cs

Author: lukebakken

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

@@ -30,10 +30,12 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using RabbitMQ.Client.Framing;
+using RabbitMQ.Client.Util;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -148,5 +150,75 @@ await MaybeEndPublisherConfirmationTracking(publisherConfirmationInfo, cancellat
                     .ConfigureAwait(false);
             }
         }
+
+        private BasicProperties? PopulateBasicPropertiesHeaders<TProperties>(TProperties basicProperties,
+            Activity? sendActivity, ulong publishSequenceNumber)
+            where TProperties : IReadOnlyBasicProperties, IAmqpHeader
+        {
+            /*
+             * Note: there is nothing to do in this method if *both* of these
+             * conditions are true:
+             *
+             * sendActivity is null - there is no activity to add as a header
+             * publisher confirmations are NOT enabled
+             */
+            if (sendActivity is null && !_publisherConfirmationsEnabled)
+            {
+                return null;
+            }
+
+            bool newHeaders = false;
+            IDictionary<string, object?>? headers = basicProperties.Headers;
+            if (headers is null)
+            {
+                headers = new Dictionary<string, object?>();
+                newHeaders = true;
+            }
+            MaybeAddActivityToHeaders(headers, basicProperties.CorrelationId, sendActivity);
+            MaybeAddPublishSequenceNumberToHeaders(headers);
+
+            switch (basicProperties)
+            {
+                case BasicProperties writableProperties:
+                    if (newHeaders)
+                    {
+                        writableProperties.Headers = headers;
+                    }
+                    return null;
+                case EmptyBasicProperty:
+                    return new BasicProperties { Headers = headers };
+                default:
+                    return new BasicProperties(basicProperties) { Headers = headers };
+            }
+
+            void MaybeAddActivityToHeaders(IDictionary<string, object?> headers,
+                string? correlationId, Activity? sendActivity)
+            {
+                if (sendActivity is not null)
+                {
+                    // This activity is marked as recorded, so let's propagate the trace and span ids.
+                    if (sendActivity.IsAllDataRequested)
+                    {
+                        if (!string.IsNullOrEmpty(correlationId))
+                        {
+                            sendActivity.SetTag(RabbitMQActivitySource.MessageConversationId, correlationId);
+                        }
+                    }
+
+                    // Inject the ActivityContext into the message headers to propagate trace context to the receiving service.
+                    RabbitMQActivitySource.ContextInjector(sendActivity, headers);
+                }
+            }
+
+            void MaybeAddPublishSequenceNumberToHeaders(IDictionary<string, object?> headers)
+            {
+                if (_publisherConfirmationsEnabled)
+                {
+                    byte[] publishSequenceNumberBytes = new byte[8];
+                    NetworkOrderSerializer.WriteUInt64(ref publishSequenceNumberBytes.GetStart(), publishSequenceNumber);
+                    headers[Constants.PublishSequenceNumberHeader] = publishSequenceNumberBytes;
+                }
+            }
+        }
     }
 }

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

@@ -30,7 +30,6 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Buffers.Binary;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -40,13 +39,12 @@
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing;
+using RabbitMQ.Client.Util;
 
 namespace RabbitMQ.Client.Impl
 {
     internal partial class Channel : IChannel, IRecoverable
     {
-        // private readonly AsyncManualResetEvent _flowControlBlock = new AsyncManualResetEvent(true);
-
         private bool _publisherConfirmationsEnabled = false;
         private bool _publisherConfirmationTrackingEnabled = false;
         private ulong _nextPublishSeqNo = 0;
@@ -230,9 +228,10 @@ private void HandleReturn(BasicReturnEventArgs basicReturnEvent)
                 ulong publishSequenceNumber = 0;
                 IReadOnlyBasicProperties props = basicReturnEvent.BasicProperties;
                 object? maybeSeqNum = props.Headers?[Constants.PublishSequenceNumberHeader];
-                if (maybeSeqNum != null)
+                if (maybeSeqNum != null &&
+                    maybeSeqNum is byte[] seqNumBytes)
                 {
-                    publishSequenceNumber = BinaryPrimitives.ReadUInt64BigEndian((byte[])maybeSeqNum);
+                    publishSequenceNumber = NetworkOrderDeserializer.ReadUInt64(seqNumBytes);
                 }
 
                 HandleNack(publishSequenceNumber, multiple: false, isReturn: true);

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -42,7 +42,6 @@
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing;
-using RabbitMQ.Client.Util;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -1500,76 +1499,6 @@ await ModelSendAsync(in method, k.CancellationToken)
             }
         }
 
-        private BasicProperties? PopulateBasicPropertiesHeaders<TProperties>(TProperties basicProperties,
-            Activity? sendActivity, ulong publishSequenceNumber)
-            where TProperties : IReadOnlyBasicProperties, IAmqpHeader
-        {
-            /*
-             * Note: there is nothing to do in this method if *both* of these
-             * conditions are true:
-             *
-             * sendActivity is null - there is no activity to add as a header
-             * publisher confirmations are NOT enabled
-             */
-            if (sendActivity is null && !_publisherConfirmationsEnabled)
-            {
-                return null;
-            }
-
-            bool newHeaders = false;
-            IDictionary<string, object?>? headers = basicProperties.Headers;
-            if (headers is null)
-            {
-                headers = new Dictionary<string, object?>();
-                newHeaders = true;
-            }
-            MaybeAddActivityToHeaders(headers, basicProperties.CorrelationId, sendActivity);
-            MaybeAddPublishSequenceNumberToHeaders(headers);
-
-            switch (basicProperties)
-            {
-                case BasicProperties writableProperties:
-                    if (newHeaders)
-                    {
-                        writableProperties.Headers = headers;
-                    }
-                    return null;
-                case EmptyBasicProperty:
-                    return new BasicProperties { Headers = headers };
-                default:
-                    return new BasicProperties(basicProperties) { Headers = headers };
-            }
-
-            void MaybeAddActivityToHeaders(IDictionary<string, object?> headers,
-                string? correlationId, Activity? sendActivity)
-            {
-                if (sendActivity is not null)
-                {
-                    // This activity is marked as recorded, so let's propagate the trace and span ids.
-                    if (sendActivity.IsAllDataRequested)
-                    {
-                        if (!string.IsNullOrEmpty(correlationId))
-                        {
-                            sendActivity.SetTag(RabbitMQActivitySource.MessageConversationId, correlationId);
-                        }
-                    }
-
-                    // Inject the ActivityContext into the message headers to propagate trace context to the receiving service.
-                    RabbitMQActivitySource.ContextInjector(sendActivity, headers);
-                }
-            }
-
-            void MaybeAddPublishSequenceNumberToHeaders(IDictionary<string, object?> headers)
-            {
-                if (_publisherConfirmationsEnabled)
-                {
-                    byte[] publishSequenceNumberBytes = new byte[8];
-                    NetworkOrderSerializer.WriteUInt64(ref publishSequenceNumberBytes.GetStart(), publishSequenceNumber);
-                    headers[Constants.PublishSequenceNumberHeader] = publishSequenceNumberBytes;
-                }
-            }
-        }
-
         /// <summary>
         /// Returning <c>true</c> from this method means that the command was server-originated,
         /// and handled already.