Avoid unnecessary allocations (Azure#16539)

* Avoid unnecessary allocations

* Update exceptions and add live test

* fix test

Author: JoshLove-msft

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

@@ -134,7 +134,7 @@ public static BinaryData GetBody(this AmqpAnnotatedMessage message)
             {
                 return dataBody.ConvertAndFlattenData();
             }
-            throw new NotSupportedException($"{message.Body.GetType()} is not a supported message body type.");
+            throw new NotSupportedException($"{message.Body.BodyType} is not a supported message body type.");
         }
     }
 }

sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Globalization;
+using System.Text;
 using Azure.Core;
 using Azure.Core.Amqp;
 using Azure.Messaging.ServiceBus.Amqp;
@@ -33,27 +34,26 @@ public ServiceBusMessage() :
         /// </summary>
         /// <param name="body">The payload of the message as a string.</param>
         public ServiceBusMessage(string body) :
-            this(new BinaryData(body))
+            this(Encoding.UTF8.GetBytes(body))
         {
         }
 
         /// <summary>
         /// Creates a new message from the specified payload.
         /// </summary>
         /// <param name="body">The payload of the message in bytes.</param>
-        public ServiceBusMessage(ReadOnlyMemory<byte> body) :
-            this(BinaryData.FromBytes(body))
+        public ServiceBusMessage(ReadOnlyMemory<byte> body)
         {
+            AmqpMessageBody amqpBody = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body });
+            AmqpMessage = new AmqpAnnotatedMessage(amqpBody);
         }
 
         /// <summary>
         /// Creates a new message from the specified <see cref="BinaryData"/> instance.
         /// </summary>
         /// <param name="body">The payload of the message.</param>
-        public ServiceBusMessage(BinaryData body)
+        public ServiceBusMessage(BinaryData body) : this (body?.ToMemory() ?? default)
         {
-            AmqpMessageBody amqpBody = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body ?? new BinaryData(Array.Empty<byte>()) });
-            AmqpMessage = new AmqpAnnotatedMessage(amqpBody);
         }
 
         /// <summary>
@@ -63,7 +63,12 @@ public ServiceBusMessage(BinaryData body)
         public ServiceBusMessage(ServiceBusReceivedMessage receivedMessage)
         {
             Argument.AssertNotNull(receivedMessage, nameof(receivedMessage));
-            AmqpMessageBody body = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { receivedMessage.Body });
+            if (!receivedMessage.AmqpMessage.Body.TryGetData(out IEnumerable<ReadOnlyMemory<byte>> dataBody))
+            {
+                throw new NotSupportedException($"{receivedMessage.AmqpMessage.Body.BodyType} is not a supported message body type.");
+            }
+
+            AmqpMessageBody body = new AmqpMessageBody(dataBody);
             AmqpMessage = new AmqpAnnotatedMessage(body);
 
             // copy properties

sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs

@@ -61,13 +61,7 @@ internal ServiceBusReceivedMessage(): this(body: default)
         /// <summary>
         /// Gets the body of the message.
         /// </summary>
-        public BinaryData Body
-        {
-            get
-            {
-                return AmqpMessage.GetBody();
-            }
-        }
+        public BinaryData Body => AmqpMessage.GetBody();
 
         /// <summary>
         /// Gets the MessageId to identify the message.

sdk/servicebus/Azure.Messaging.ServiceBus/tests/Message/MessageLiveTests.cs

@@ -90,6 +90,23 @@ public async Task CanSendMessageWithMaxSize()
             }
         }
 
+        [Test]
+        public async Task CanSendNullBodyMessage()
+        {
+            await using (var scope = await ServiceBusScope.CreateWithQueue(enablePartitioning: false, enableSession: false))
+            {
+                var client = new ServiceBusClient(TestEnvironment.ServiceBusConnectionString);
+
+                var maxSizeMessage = new ServiceBusMessage((BinaryData)null);
+
+                await client.CreateSender(scope.QueueName).SendMessageAsync(maxSizeMessage);
+                var receiver = client.CreateReceiver(scope.QueueName);
+                var receivedMessage = await receiver.ReceiveMessageAsync();
+                Assert.IsNotNull(receivedMessage);
+                await receiver.CompleteMessageAsync(receivedMessage.LockToken);
+            }
+        }
+
         [Test]
         public async Task CreateFromReceivedMessageCopiesProperties()
         {