Amqp api updates (Azure#16796)

JoshLove-msft · web-flow · commit cab3ab6cfcbe · 2020-11-10T13:27:39.000-08:00
diff --git a/sdk/core/Azure.Core.Amqp/api/Azure.Core.Amqp.netstandard2.0.cs b/sdk/core/Azure.Core.Amqp/api/Azure.Core.Amqp.netstandard2.0.cs
@@ -18,7 +18,7 @@ public partial struct AmqpAddress : System.IEquatable<Azure.Core.Amqp.AmqpAddres
     }
     public partial class AmqpAnnotatedMessage
     {
-        public AmqpAnnotatedMessage(System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> dataBody) { }
+        public AmqpAnnotatedMessage(Azure.Core.Amqp.AmqpMessageBody body) { }
         public System.Collections.Generic.IDictionary<string, object> ApplicationProperties { get { throw null; } }
         public Azure.Core.Amqp.AmqpMessageBody Body { get { throw null; } set { } }
         public System.Collections.Generic.IDictionary<string, object> DeliveryAnnotations { get { throw null; } }
@@ -27,16 +27,11 @@ public AmqpAnnotatedMessage(System.Collections.Generic.IEnumerable<System.ReadOn
         public System.Collections.Generic.IDictionary<string, object> MessageAnnotations { get { throw null; } }
         public Azure.Core.Amqp.AmqpMessageProperties Properties { get { throw null; } }
     }
-    public partial class AmqpDataMessageBody : Azure.Core.Amqp.AmqpMessageBody
+    public partial class AmqpMessageBody
     {
-        public AmqpDataMessageBody(System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> data) { }
-        public override Azure.Core.Amqp.AmqpMessageBodyType BodyType { get { throw null; } }
-        public virtual System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> Data { get { throw null; } }
-    }
-    public abstract partial class AmqpMessageBody
-    {
-        protected AmqpMessageBody() { }
-        public abstract Azure.Core.Amqp.AmqpMessageBodyType BodyType { get; }
+        public AmqpMessageBody(System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> data) { }
+        public Azure.Core.Amqp.AmqpMessageBodyType BodyType { get { throw null; } }
+        public bool TryGetData(out System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>>? data) { throw null; }
     }
     public enum AmqpMessageBodyType
     {
diff --git a/sdk/core/Azure.Core.Amqp/src/AmqpAnnotatedMessage.cs b/sdk/core/Azure.Core.Amqp/src/AmqpAnnotatedMessage.cs
@@ -15,12 +15,12 @@ public class AmqpAnnotatedMessage
         /// <summary>
         /// Initializes a new Data body <see cref="AmqpAnnotatedMessage"/>.
         /// </summary>
-        /// <param name="dataBody">The data sections comprising the message body.
+        /// <param name="body">The data sections comprising the message body.
         /// <seealso href="http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-data"/>
         /// </param>
-        public AmqpAnnotatedMessage(IEnumerable<ReadOnlyMemory<byte>> dataBody)
+        public AmqpAnnotatedMessage(AmqpMessageBody body)
         {
-            Body = new AmqpDataMessageBody(dataBody);
+            Body = body;
         }
 
         /// <summary>
diff --git a/sdk/core/Azure.Core.Amqp/src/AmqpDataMessageBody.cs b/sdk/core/Azure.Core.Amqp/src/AmqpDataMessageBody.cs
diff --git a/sdk/core/Azure.Core.Amqp/src/AmqpMessageBody.cs b/sdk/core/Azure.Core.Amqp/src/AmqpMessageBody.cs
@@ -1,17 +1,53 @@
 ﻿// Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
 namespace Azure.Core.Amqp
 {
     /// <summary>
     /// Represents an AMQP message body.
     /// <seealso href="http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format" />
     /// </summary>
-    public abstract class AmqpMessageBody
+    public class AmqpMessageBody
     {
+        /// <summary>
+        /// The data sections for the AMQP message body.
+        /// </summary>
+        private readonly IEnumerable<ReadOnlyMemory<byte>> _data;
+
         /// <summary>
         /// Gets the type of the message body.
         /// </summary>
-        public abstract AmqpMessageBodyType BodyType { get; }
+        public AmqpMessageBodyType BodyType { get; }
+
+        /// <summary>
+        /// Initializes a new <see cref="AmqpMessageBody"/> instance with the
+        /// passed in data sections.
+        /// </summary>
+        /// <param name="data">The data sections.</param>
+        public AmqpMessageBody(IEnumerable<ReadOnlyMemory<byte>> data)
+        {
+            _data = data ?? Enumerable.Empty<ReadOnlyMemory<byte>>();
+            BodyType = AmqpMessageBodyType.Data;
+        }
+
+        /// <summary>
+        /// Try to get the data sections for the AMQP message body.
+        /// </summary>
+        /// <param name="data"></param>
+        /// <returns></returns>
+        public bool TryGetData(out IEnumerable<ReadOnlyMemory<byte>>? data)
+        {
+            if (BodyType == AmqpMessageBodyType.Data)
+            {
+                data = _data;
+                return true;
+            }
+            data = null;
+            return false;
+        }
     }
 }
diff --git a/sdk/core/Azure.Core.Amqp/tests/AmqpAnnotatedMessageTests.cs b/sdk/core/Azure.Core.Amqp/tests/AmqpAnnotatedMessageTests.cs
@@ -14,7 +14,7 @@ public class AmqpAnnotatedMessageTests
         [Test]
         public void CanCreateAnnotatedMessage()
         {
-            var message = new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { Encoding.UTF8.GetBytes("some data") });
+            var message = new AmqpAnnotatedMessage(new AmqpMessageBody(new ReadOnlyMemory<byte>[] { Encoding.UTF8.GetBytes("some data") }));
             message.ApplicationProperties.Add("applicationKey", "applicationValue");
             message.DeliveryAnnotations.Add("deliveryKey", "deliveryValue");
             message.MessageAnnotations.Add("messageKey", "messageValue");
@@ -40,7 +40,8 @@ public void CanCreateAnnotatedMessage()
             message.Properties.UserId = Encoding.UTF8.GetBytes("userId");
 
             Assert.AreEqual(AmqpMessageBodyType.Data, message.Body.BodyType);
-            Assert.AreEqual("some data", Encoding.UTF8.GetString(((AmqpDataMessageBody)message.Body).Data.First().ToArray()));
+            Assert.IsTrue(message.Body.TryGetData(out IEnumerable<ReadOnlyMemory<byte>> body));
+            Assert.AreEqual("some data", Encoding.UTF8.GetString(body.First().ToArray()));
             Assert.AreEqual("applicationValue", message.ApplicationProperties["applicationKey"]);
             Assert.AreEqual("deliveryValue", message.DeliveryAnnotations["deliveryKey"]);
             Assert.AreEqual("messageValue", message.MessageAnnotations["messageKey"]);
diff --git a/sdk/core/Azure.Core.Amqp/tests/AmqpDataMessageBodyTests.cs b/sdk/core/Azure.Core.Amqp/tests/AmqpDataMessageBodyTests.cs
@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 using System;
+using System.Collections.Generic;
 using NUnit.Framework;
 
 namespace Azure.Core.Amqp.Tests
@@ -11,8 +12,15 @@ public class AmqpDataMessageBodyTests
         [Test]
         public void CanCreateDataBody()
         {
-            var body = new AmqpDataMessageBody(Array.Empty<ReadOnlyMemory<byte>>());
+            var body = new AmqpMessageBody(Array.Empty<ReadOnlyMemory<byte>>());
             Assert.AreEqual(AmqpMessageBodyType.Data, body.BodyType);
+            Assert.IsTrue(body.TryGetData(out var data));
+            Assert.NotNull(data);
+
+            body = new AmqpMessageBody(null);
+            Assert.AreEqual(AmqpMessageBodyType.Data, body.BodyType);
+            Assert.IsTrue(body.TryGetData(out data));
+            Assert.NotNull(data);
         }
     }
 }
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpMessageConverter.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpMessageConverter.cs
@@ -214,11 +214,11 @@ public static ServiceBusReceivedMessage AmqpMessageToSBMessage(AmqpMessage amqpM
 
             if ((amqpMessage.BodyType & SectionFlag.Data) != 0 && amqpMessage.DataBody != null)
             {
-                annotatedMessage = new AmqpAnnotatedMessage(amqpMessage.GetDataViaDataBody());
+                annotatedMessage = new AmqpAnnotatedMessage(new AmqpMessageBody(amqpMessage.GetDataViaDataBody()));
             }
             else
             {
-                annotatedMessage = new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { Array.Empty<byte>() });
+                annotatedMessage = new AmqpAnnotatedMessage(new AmqpMessageBody(Enumerable.Empty<ReadOnlyMemory<byte>>()));
             }
             ServiceBusReceivedMessage sbMessage = new ServiceBusReceivedMessage(annotatedMessage);
 
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpMessageExtensions.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpMessageExtensions.cs
@@ -15,9 +15,9 @@ internal static class AmqpMessageExtensions
     {
         public static AmqpMessage ToAmqpMessage(this ServiceBusMessage message)
         {
-            if (message.AmqpMessage.Body is AmqpDataMessageBody dataBody)
+            if (message.AmqpMessage.Body.TryGetData(out IEnumerable<ReadOnlyMemory<byte>> dataBody))
             {
-                return AmqpMessage.Create(dataBody.Data.AsAmqpData());
+                return AmqpMessage.Create(dataBody.AsAmqpData());
             }
             throw new NotSupportedException($"{message.AmqpMessage.Body.GetType()} is not a supported message body type.");
         }
@@ -130,9 +130,9 @@ public static DateTimeOffset GetScheduledEnqueueTime(this AmqpAnnotatedMessage m
 
         public static BinaryData GetBody(this AmqpAnnotatedMessage message)
         {
-            if (message.Body is AmqpDataMessageBody dataBody)
+            if (message.Body.TryGetData(out IEnumerable<ReadOnlyMemory<byte>> dataBody))
             {
-                return dataBody.Data.ConvertAndFlattenData();
+                return dataBody.ConvertAndFlattenData();
             }
             throw new NotSupportedException($"{message.Body.GetType()} is not a supported message body type.");
         }
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Azure.Messaging.ServiceBus.csproj b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Azure.Messaging.ServiceBus.csproj
@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+﻿<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
     <Description>Azure Service Bus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. This client library allows for both sending and receiving messages using Azure Service Bus. For more information about Service Bus, see https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview</Description>
     <Version>7.0.0-preview.10</Version>
@@ -12,7 +12,6 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Azure.Core.Amqp" />
     <PackageReference Include="Microsoft.Azure.Amqp" />
     <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" />
     <!--This can be removed once this dependency is added to Core in the next release.-->
@@ -36,6 +35,9 @@
     <Compile Include="$(AzureCoreSharedSources)PageResponseEnumerator.cs" Link="SharedSource\Azure.Core\PageResponseEnumerator.cs" />
     <Compile Include="$(AzureCoreSharedSources)AzureResourceProviderNamespaceAttribute.cs" Link="SharedSource\Azure.Core\AzureResourceProviderNamespaceAttribute.cs" />
   </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\core\Azure.Core.Amqp\src\Azure.Core.Amqp.csproj" />
+  </ItemGroup>
   <ItemGroup>
     <Compile Update="Resources.Designer.cs">
       <DesignTime>True</DesignTime>
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Plugins/ServiceBusPlugin.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Plugins/ServiceBusPlugin.cs
@@ -39,7 +39,7 @@ public virtual ValueTask AfterMessageReceiveAsync(ServiceBusReceivedMessage mess
 #pragma warning disable CA1822 // Mark members as static
         protected void SetBody(ServiceBusReceivedMessage message, BinaryData body)
         {
-            message.AmqpMessage.Body = new AmqpDataMessageBody(new ReadOnlyMemory<byte>[] { body });
+            message.AmqpMessage.Body = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body });
         }
 
         /// <summary>
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs
@@ -52,7 +52,8 @@ public ServiceBusMessage(ReadOnlyMemory<byte> body) :
         /// <param name="body">The payload of the message.</param>
         public ServiceBusMessage(BinaryData body)
         {
-            AmqpMessage = new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { body ?? new BinaryData(Array.Empty<byte>()) });
+            AmqpMessageBody amqpBody = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body ?? new BinaryData(Array.Empty<byte>()) });
+            AmqpMessage = new AmqpAnnotatedMessage(amqpBody);
         }
 
         /// <summary>
@@ -62,7 +63,8 @@ public ServiceBusMessage(BinaryData body)
         public ServiceBusMessage(ServiceBusReceivedMessage receivedMessage)
         {
             Argument.AssertNotNull(receivedMessage, nameof(receivedMessage));
-            AmqpMessage = new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { receivedMessage.Body });
+            AmqpMessageBody body = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { receivedMessage.Body });
+            AmqpMessage = new AmqpAnnotatedMessage(body);
 
             // copy properties
             AmqpMessageProperties properties = AmqpMessage.Properties;
@@ -133,7 +135,7 @@ public BinaryData Body
             get => AmqpMessage.GetBody();
             set
             {
-                AmqpMessage.Body = new AmqpDataMessageBody(new ReadOnlyMemory<byte>[] { value });
+                AmqpMessage.Body = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { value });
             }
         }
 
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusModelFactory.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusModelFactory.cs
@@ -43,7 +43,7 @@ public static ServiceBusReceivedMessage ServiceBusReceivedMessage(
             long enqueuedSequenceNumber = default,
             DateTimeOffset enqueuedTime = default)
         {
-            var amqpMessage = new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { body });
+            var amqpMessage = new AmqpAnnotatedMessage(new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body }));
             amqpMessage.Properties.CorrelationId = new AmqpMessageId(correlationId);
             amqpMessage.Properties.Subject = subject;
             amqpMessage.Properties.To = new AmqpAddress(to);
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs
@@ -22,7 +22,7 @@ public class ServiceBusReceivedMessage
         /// </summary>
         /// <param name="body">The payload of the message represented as bytes.</param>
         internal ServiceBusReceivedMessage(ReadOnlyMemory<byte> body)
-            : this(new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { body }))
+            : this(new AmqpAnnotatedMessage(new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body })))
         {
         }
 
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Message/MessageLiveTests.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Message/MessageLiveTests.cs
@@ -198,21 +198,22 @@ public async Task CanSendMultipleDataSections()
 
                 var msg = new ServiceBusMessage();
                 var amqp = new AmqpAnnotatedMessage(
+                    new AmqpMessageBody(
                     new ReadOnlyMemory<byte>[]
                     {
                         new ReadOnlyMemory<byte>(GetRandomBuffer(100)),
                         new ReadOnlyMemory<byte>(GetRandomBuffer(100))
-                    });
+                    }));
                 msg.AmqpMessage = amqp;
 
                 await sender.SendMessageAsync(msg);
 
                 var receiver = client.CreateReceiver(scope.QueueName);
                 var received = await receiver.ReceiveMessageAsync();
-                var receivedData = ((AmqpDataMessageBody)received.GetRawMessage().Body).Data;
+                received.GetRawMessage().Body.TryGetData(out var receivedData);
                 var bodyEnum = receivedData.GetEnumerator();
                 int ct = 0;
-                var sentData = ((AmqpDataMessageBody)msg.GetRawMessage().Body).Data;
+                msg.GetRawMessage().Body.TryGetData(out var sentData);
 
                 foreach (ReadOnlyMemory<byte> data in sentData)
                 {
@@ -239,7 +240,7 @@ public async Task CanSetMessageId()
                 var client = new ServiceBusClient(TestEnvironment.ServiceBusConnectionString);
                 var sender = client.CreateSender(scope.QueueName);
                 var msg = new ServiceBusMessage();
-                msg.GetRawMessage().Body = new AmqpDataMessageBody(new ReadOnlyMemory<byte>[]
+                msg.GetRawMessage().Body = new AmqpMessageBody(new ReadOnlyMemory<byte>[]
                     {
                         new ReadOnlyMemory<byte>(GetRandomBuffer(100)),
                         new ReadOnlyMemory<byte>(GetRandomBuffer(100))

Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,7 @@ public partial struct AmqpAddress : System.IEquatable<Azure.Core.Amqp.AmqpAddres`
`18`	`18`	`}`
`19`	`19`	`public partial class AmqpAnnotatedMessage`
`20`	`20`	`{`
`21`		`- public AmqpAnnotatedMessage(System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> dataBody) { }`
	`21`	`+ public AmqpAnnotatedMessage(Azure.Core.Amqp.AmqpMessageBody body) { }`
`22`	`22`	`public System.Collections.Generic.IDictionary<string, object> ApplicationProperties { get { throw null; } }`
`23`	`23`	`public Azure.Core.Amqp.AmqpMessageBody Body { get { throw null; } set { } }`
`24`	`24`	`public System.Collections.Generic.IDictionary<string, object> DeliveryAnnotations { get { throw null; } }`
`@@ -27,16 +27,11 @@ public AmqpAnnotatedMessage(System.Collections.Generic.IEnumerable<System.ReadOn`
`27`	`27`	`public System.Collections.Generic.IDictionary<string, object> MessageAnnotations { get { throw null; } }`
`28`	`28`	`public Azure.Core.Amqp.AmqpMessageProperties Properties { get { throw null; } }`
`29`	`29`	`}`
`30`		`- public partial class AmqpDataMessageBody : Azure.Core.Amqp.AmqpMessageBody`
	`30`	`+ public partial class AmqpMessageBody`
`31`	`31`	`{`
`32`		`- public AmqpDataMessageBody(System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> data) { }`
`33`		`- public override Azure.Core.Amqp.AmqpMessageBodyType BodyType { get { throw null; } }`
`34`		`- public virtual System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> Data { get { throw null; } }`
`35`		`- }`
`36`		`- public abstract partial class AmqpMessageBody`
`37`		`- {`
`38`		`- protected AmqpMessageBody() { }`
`39`		`- public abstract Azure.Core.Amqp.AmqpMessageBodyType BodyType { get; }`
	`32`	`+ public AmqpMessageBody(System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>> data) { }`
	`33`	`+ public Azure.Core.Amqp.AmqpMessageBodyType BodyType { get { throw null; } }`
	`34`	`+ public bool TryGetData(out System.Collections.Generic.IEnumerable<System.ReadOnlyMemory<byte>>? data) { throw null; }`
`40`	`35`	`}`
`41`	`36`	`public enum AmqpMessageBodyType`
`42`	`37`	`{`
Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`	`// Licensed under the MIT License.`
`3`	`3`
`4`	`4`	`using System;`
	`5`	`+using System.Collections.Generic;`
`5`	`6`	`using NUnit.Framework;`
`6`	`7`
`7`	`8`	`namespace Azure.Core.Amqp.Tests`
`@@ -11,8 +12,15 @@ public class AmqpDataMessageBodyTests`
`11`	`12`	`[Test]`
`12`	`13`	`public void CanCreateDataBody()`
`13`	`14`	`{`
`14`		`- var body = new AmqpDataMessageBody(Array.Empty<ReadOnlyMemory<byte>>());`
	`15`	`+ var body = new AmqpMessageBody(Array.Empty<ReadOnlyMemory<byte>>());`
`15`	`16`	`Assert.AreEqual(AmqpMessageBodyType.Data, body.BodyType);`
	`17`	`+ Assert.IsTrue(body.TryGetData(out var data));`
	`18`	`+ Assert.NotNull(data);`
	`19`	`+`
	`20`	`+ body = new AmqpMessageBody(null);`
	`21`	`+ Assert.AreEqual(AmqpMessageBodyType.Data, body.BodyType);`
	`22`	`+ Assert.IsTrue(body.TryGetData(out data));`
	`23`	`+ Assert.NotNull(data);`
`16`	`24`	`}`
`17`	`25`	`}`
`18`	`26`	`}`
Original file line number	Diff line number	Diff line change
`@@ -214,11 +214,11 @@ public static ServiceBusReceivedMessage AmqpMessageToSBMessage(AmqpMessage amqpM`
`214`	`214`
`215`	`215`	`if ((amqpMessage.BodyType & SectionFlag.Data) != 0 && amqpMessage.DataBody != null)`
`216`	`216`	`{`
`217`		`- annotatedMessage = new AmqpAnnotatedMessage(amqpMessage.GetDataViaDataBody());`
	`217`	`+ annotatedMessage = new AmqpAnnotatedMessage(new AmqpMessageBody(amqpMessage.GetDataViaDataBody()));`
`218`	`218`	`}`
`219`	`219`	`else`
`220`	`220`	`{`
`221`		`- annotatedMessage = new AmqpAnnotatedMessage(new ReadOnlyMemory<byte>[] { Array.Empty<byte>() });`
	`221`	`+ annotatedMessage = new AmqpAnnotatedMessage(new AmqpMessageBody(Enumerable.Empty<ReadOnlyMemory<byte>>()));`
`222`	`222`	`}`
`223`	`223`	`ServiceBusReceivedMessage sbMessage = new ServiceBusReceivedMessage(annotatedMessage);`
`224`	`224`
Original file line number	Diff line number	Diff line change
`@@ -15,9 +15,9 @@ internal static class AmqpMessageExtensions`
`15`	`15`	`{`
`16`	`16`	`public static AmqpMessage ToAmqpMessage(this ServiceBusMessage message)`
`17`	`17`	`{`
`18`		`- if (message.AmqpMessage.Body is AmqpDataMessageBody dataBody)`
	`18`	`+ if (message.AmqpMessage.Body.TryGetData(out IEnumerable<ReadOnlyMemory<byte>> dataBody))`
`19`	`19`	`{`
`20`		`- return AmqpMessage.Create(dataBody.Data.AsAmqpData());`
	`20`	`+ return AmqpMessage.Create(dataBody.AsAmqpData());`
`21`	`21`	`}`
`22`	`22`	`throw new NotSupportedException($"{message.AmqpMessage.Body.GetType()} is not a supported message body type.");`
`23`	`23`	`}`
`@@ -130,9 +130,9 @@ public static DateTimeOffset GetScheduledEnqueueTime(this AmqpAnnotatedMessage m`
`130`	`130`
`131`	`131`	`public static BinaryData GetBody(this AmqpAnnotatedMessage message)`
`132`	`132`	`{`
`133`		`- if (message.Body is AmqpDataMessageBody dataBody)`
	`133`	`+ if (message.Body.TryGetData(out IEnumerable<ReadOnlyMemory<byte>> dataBody))`
`134`	`134`	`{`
`135`		`- return dataBody.Data.ConvertAndFlattenData();`
	`135`	`+ return dataBody.ConvertAndFlattenData();`
`136`	`136`	`}`
`137`	`137`	`throw new NotSupportedException($"{message.Body.GetType()} is not a supported message body type.");`
`138`	`138`	`}`
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ public virtual ValueTask AfterMessageReceiveAsync(ServiceBusReceivedMessage mess`
`39`	`39`	`#pragma warning disable CA1822 // Mark members as static`
`40`	`40`	`protected void SetBody(ServiceBusReceivedMessage message, BinaryData body)`
`41`	`41`	`{`
`42`		`- message.AmqpMessage.Body = new AmqpDataMessageBody(new ReadOnlyMemory<byte>[] { body });`
	`42`	`+ message.AmqpMessage.Body = new AmqpMessageBody(new ReadOnlyMemory<byte>[] { body });`
`43`	`43`	`}`
`44`	`44`
`45`	`45`	`/// <summary>`