[ServiceBus] No longer create a copy of the array in the message converter for the default and stream case (Azure#29946)

* No longer create a copy of the array in the message converter for the default case and when handling streams

* Apply same principles to StreamToBytes

* Remove unneeded method

* Align StreamToBytes

* Collapse stream processing

* Add proposal for the changelog

Author: danielmarbach

sdk/servicebus/Azure.Messaging.ServiceBus/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Other Changes
 
+- Reduced memory allocations when converting messages into the underlying AMQP primitives. _(A community contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_
+
 ## 7.9.0 (2022-07-11)
 
 ### Features Added

sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpMessageConverter.cs

@@ -145,28 +145,34 @@ private static AmqpMessage BuildAmqpBatchFromMessages(
         ///
         private static ArraySegment<byte> ReadStreamToArraySegment(Stream stream)
         {
-            if (stream == null)
-            {
-                return new ArraySegment<byte>();
-            }
-
             switch (stream)
             {
+                case { Length: < 1 }:
+                    return default;
+
                 case BufferListStream bufferListStream:
                     return bufferListStream.ReadBytes((int)stream.Length);
 
                 case MemoryStream memStreamSource:
                 {
                     using var memStreamCopy = new MemoryStream((int)(memStreamSource.Length - memStreamSource.Position));
                     memStreamSource.CopyTo(memStreamCopy, StreamBufferSizeInBytes);
-                    return new ArraySegment<byte>(memStreamCopy.ToArray());
+                    if (!memStreamCopy.TryGetBuffer(out ArraySegment<byte> segment))
+                    {
+                        segment = new ArraySegment<byte>(memStreamCopy.ToArray());
+                    }
+                    return segment;
                 }
 
                 default:
                 {
-                    using var memStream = new MemoryStream(StreamBufferSizeInBytes);
-                    stream.CopyTo(memStream, StreamBufferSizeInBytes);
-                    return new ArraySegment<byte>(memStream.ToArray());
+                    using var memStreamCopy = new MemoryStream(StreamBufferSizeInBytes);
+                    stream.CopyTo(memStreamCopy, StreamBufferSizeInBytes);
+                    if (!memStreamCopy.TryGetBuffer(out ArraySegment<byte> segment))
+                    {
+                        segment = new ArraySegment<byte>(memStreamCopy.ToArray());
+                    }
+                    return segment;
                 }
             }
         }
@@ -689,7 +695,7 @@ internal static bool TryGetAmqpObjectFromNetObject(object netObject, MappingType
                 case PropertyValueType.Stream:
                     if (mappingType == MappingType.ApplicationProperty)
                     {
-                        amqpObject = StreamToBytes((Stream)netObject);
+                        amqpObject = ReadStreamToArraySegment((Stream)netObject);
                     }
                     break;
                 case PropertyValueType.Uri:
@@ -706,7 +712,7 @@ internal static bool TryGetAmqpObjectFromNetObject(object netObject, MappingType
                     {
                         if (mappingType == MappingType.ApplicationProperty)
                         {
-                            amqpObject = StreamToBytes(netObjectAsStream);
+                            amqpObject = ReadStreamToArraySegment(netObjectAsStream);
                         }
                     }
                     else if (mappingType == MappingType.ApplicationProperty)
@@ -822,33 +828,6 @@ private static bool TryGetNetObjectFromAmqpObject(object amqpObject, MappingType
             return netObject != null;
         }
 
-        private static ArraySegment<byte> StreamToBytes(Stream stream)
-        {
-            ArraySegment<byte> buffer;
-            if (stream == null || stream.Length < 1)
-            {
-                buffer = default;
-            }
-            else
-            {
-                using (var memoryStream = new MemoryStream(512))
-                {
-                    stream.CopyTo(memoryStream, 512);
-                    buffer = new ArraySegment<byte>(memoryStream.ToArray());
-                }
-            }
-
-            return buffer;
-        }
-
-        private static Data ToData(AmqpMessage message)
-        {
-            ArraySegment<byte>[] payload = message.GetPayload();
-            var buffer = new BufferListStream(payload);
-            ArraySegment<byte> value = buffer.ReadBytes((int)buffer.Length);
-            return new Data { Value = value };
-        }
-
         internal static AmqpMap GetSqlRuleFilterMap(SqlRuleFilter sqlRuleFilter)
         {
             var amqpFilterMap = new AmqpMap