Update RabbitMQ client version to 6.2.2

WebJobs.Extensions.RabbitMQ.sln

@@ -1,7 +1,7 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 16
-VisualStudioVersion = 16.0.29020.237
+# Visual Studio Version 17
+VisualStudioVersion = 17.0.31912.275
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WebJobs.Extensions.RabbitMQ", "src\WebJobs.Extensions.RabbitMQ.csproj", "{9DB4F409-0611-42FA-A64E-B17C23A406D3}"
 EndProject

azure-pipelines.yml

@@ -36,7 +36,7 @@ jobs:
         displayName: Acquire .NET SDK
         inputs:
           packageType: sdk
-          version: 2.x
+          version: 6.x
           performMultiLevelLookup: true
 
       - task: DotNetCoreCLI@2

src/Bindings/RabbitMQAsyncCollector.cs

@@ -5,10 +5,11 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
+using RabbitMQ.Client;
 
 namespace Microsoft.Azure.WebJobs.Extensions.RabbitMQ
 {
-    internal class RabbitMQAsyncCollector : IAsyncCollector<byte[]>
+    internal class RabbitMQAsyncCollector : IAsyncCollector<ReadOnlyMemory<byte>>
     {
         private readonly RabbitMQContext _context;
         private readonly ILogger _logger;
@@ -23,7 +24,7 @@ public RabbitMQAsyncCollector(RabbitMQContext context, ILogger logger)
             }
         }
 
-        public Task AddAsync(byte[] message, CancellationToken cancellationToken = default)
+        public Task AddAsync(ReadOnlyMemory<byte> message, CancellationToken cancellationToken = default)
         {
             _logger.LogDebug($"Adding message to batch for publishing...");
 

src/Config/PocoToBytesConverter.cs

@@ -7,9 +7,9 @@
 
 namespace Microsoft.Azure.WebJobs.Extensions.RabbitMQ
 {
-    internal class PocoToBytesConverter<T> : IConverter<T, byte[]>
+    internal class PocoToBytesConverter<T> : IConverter<T, ReadOnlyMemory<byte>>
     {
-        public byte[] Convert(T input)
+        public ReadOnlyMemory<byte> Convert(T input)
         {
             if (input == null)
             {

src/Config/RabbitMQExtensionConfigProvider.cs

@@ -44,13 +44,14 @@ public void Initialize(ExtensionConfigContext context)
 
             var rule = context.AddBindingRule<RabbitMQAttribute>();
             rule.AddValidator(ValidateBinding);
-            rule.BindToCollector<byte[]>((attr) =>
+            rule.BindToCollector<ReadOnlyMemory<byte>>((attr) =>
             {
                 return new RabbitMQAsyncCollector(CreateContext(attr), _logger);
             });
             rule.BindToInput<IModel>(new RabbitMQClientBuilder(this, _options));
-            rule.AddConverter<string, byte[]>(msg => Encoding.UTF8.GetBytes(msg));
-            rule.AddOpenConverter<OpenType.Poco, byte[]>(typeof(PocoToBytesConverter<>));
+            rule.AddConverter<string, ReadOnlyMemory<byte>>(arg => Encoding.UTF8.GetBytes(arg));
+            rule.AddConverter<byte[], ReadOnlyMemory<byte>>(arg => arg);
+            rule.AddOpenConverter<OpenType.Poco, ReadOnlyMemory<byte>>(typeof(PocoToBytesConverter<>));
 
             var triggerRule = context.AddBindingRule<RabbitMQTriggerAttribute>();
 

src/Trigger/BasicDeliverEventArgsValueProvider.cs

@@ -28,10 +28,14 @@ public Task<object> GetValueAsync()
             {
                 return Task.FromResult<object>(_input);
             }
-            else if (Type.Equals(typeof(byte[])))
+            else if (Type.Equals(typeof(ReadOnlyMemory<byte>)))
             {
                 return Task.FromResult<object>(_input.Body);
             }
+            else if (Type.Equals(typeof(byte[])))
+            {
+                return Task.FromResult<object>(_input.Body.ToArray());
+            }
 
             string inputValue = ToInvokeString();
             if (Type.Equals(typeof(string)))
@@ -55,7 +59,7 @@ public Task<object> GetValueAsync()
 
         public string ToInvokeString()
         {
-            return Encoding.UTF8.GetString(_input.Body);
+            return Encoding.UTF8.GetString(_input.Body.ToArray());
         }
     }
 }

src/Trigger/IRabbitMQModel.cs

@@ -1,6 +1,7 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
+using System;
 using System.Collections.Generic;
 using RabbitMQ.Client;
 
@@ -26,7 +27,7 @@ public interface IRabbitMQModel
 
         void BasicReject(ulong deliveryTag, bool requeue);
 
-        void BasicPublish(string exchange, string routingKey, IBasicProperties basicProperties, byte[] body);
+        void BasicPublish(string exchange, string routingKey, IBasicProperties basicProperties, ReadOnlyMemory<byte> body);
 
         void BasicCancel(string consumerTag);
 

src/Trigger/RabbitMQModel.cs

@@ -1,6 +1,7 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
+using System;
 using System.Collections.Generic;
 using RabbitMQ.Client;
 
@@ -57,7 +58,7 @@ public void BasicReject(ulong deliveryTag, bool requeue)
             _model.BasicReject(deliveryTag, requeue);
         }
 
-        public void BasicPublish(string exchange, string routingKey, IBasicProperties basicProperties, byte[] body)
+        public void BasicPublish(string exchange, string routingKey, IBasicProperties basicProperties, ReadOnlyMemory<byte> body)
         {
             _model.BasicPublish(exchange, routingKey, basicProperties, body);
         }

src/Trigger/RabbitMQTriggerBinding.cs

@@ -81,7 +81,7 @@ internal static IReadOnlyDictionary<string, Type> CreateBindingDataContract()
             contract.Add("Exchange", typeof(string));
             contract.Add("RoutingKey", typeof(string));
             contract.Add("BasicProperties", typeof(IBasicProperties));
-            contract.Add("Body", typeof(byte[]));
+            contract.Add("Body", typeof(ReadOnlyMemory<byte>));
 
             return contract;
         }

src/WebJobs.Extensions.RabbitMQ.csproj

@@ -46,7 +46,7 @@
 
   <ItemGroup>
     <PackageReference Include="Microsoft.Azure.WebJobs" Version="3.0.14" />
-    <PackageReference Include="RabbitMQ.Client" Version="5.1.2" />
+    <PackageReference Include="RabbitMQ.Client" Version="6.2.2" />
     <PackageReference Include="StyleCop.Analyzers" Version="1.1.118">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

test/WebJobs.Extensions.RabbitMQ.Samples/Program.cs

@@ -26,7 +26,8 @@ public static async Task Main(string[] args)
                {
                    webJobsBuilder
                    .AddAzureStorageCoreServices()
-                   .AddAzureStorage()
+                   .AddAzureStorageBlobs()
+                   .AddAzureStorageQueues()
                    .AddRabbitMQ()
                    .AddTimers();
                })

test/WebJobs.Extensions.RabbitMQ.Samples/RabbitMQSamples.cs

@@ -106,7 +106,7 @@ public static void RabbitMQTrigger_BasicDeliverEventArgs(
             [RabbitMQTrigger("queue")] BasicDeliverEventArgs args,
             ILogger logger)
         {
-            logger.LogInformation($"RabbitMQ queue trigger function processed message: {Encoding.UTF8.GetString(args.Body)}");
+            logger.LogInformation($"RabbitMQ queue trigger function processed message: {Encoding.UTF8.GetString(args.Body.ToArray())}");
         }
 
         // This sample should fail when running a console app that sends out a message incorrectly formatted.

test/WebJobs.Extensions.RabbitMQ.Samples/WebJobs.Extensions.RabbitMQ.Samples.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFramework>netcoreapp2.2</TargetFramework>
+    <TargetFramework>net6.0</TargetFramework>
     <LangVersion>latest</LangVersion>
   </PropertyGroup>
 
@@ -15,11 +15,9 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Azure.WebJobs" Version="3.0.14" />
-    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions" Version="3.0.5" />
-    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Storage" Version="3.0.10" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="2.2.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="2.2.0" />
+    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions" Version="4.0.1" />
+    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Storage" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="6.0.0" />
     <PackageReference Include="StyleCop.Analyzers" Version="1.1.118">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

test/WebJobs.Extensions.RabbitMQ.Tests/BasicDeliverEventArgsValueProviderTests.cs

@@ -7,10 +7,8 @@
 using Microsoft.Azure.WebJobs.Extensions.RabbitMQ;
 using Newtonsoft.Json;
 using RabbitMQ.Client.Events;
-using RabbitMQ.Client.Framing;
 using Xunit;
 
-
 namespace WebJobs.Extensions.RabbitMQ.Tests
 {
     public class BasicDeliverEventArgsValueProviderTests
@@ -56,7 +54,7 @@ public async Task BasicDeliverEventArgs_NoConversion_Succeeds()
         {
             string expectedString = "someString";
             byte[] stringInBytes = Encoding.UTF8.GetBytes(expectedString);
-            BasicDeliverEventArgs exceptedObject = new BasicDeliverEventArgs("tag", 1, false, "", "queue", new BasicProperties(), stringInBytes);
+            BasicDeliverEventArgs exceptedObject = new BasicDeliverEventArgs("tag", 1, false, "", "queue", null, stringInBytes);
             BasicDeliverEventArgsValueProvider testValueProvider = new BasicDeliverEventArgsValueProvider(exceptedObject, typeof(BasicDeliverEventArgs));
 
             BasicDeliverEventArgs actualObject = (BasicDeliverEventArgs)await testValueProvider.GetValueAsync();
@@ -71,7 +69,7 @@ public async Task ByteArray_Conversion_Succeeds()
         {
             string expectedString = "someString";
             byte[] stringInBytes = Encoding.UTF8.GetBytes(expectedString);
-            BasicDeliverEventArgs exceptedObject = new BasicDeliverEventArgs("tag", 1, false, "", "queue", new BasicProperties(), stringInBytes);
+            BasicDeliverEventArgs exceptedObject = new BasicDeliverEventArgs("tag", 1, false, "", "queue", null, stringInBytes);
             BasicDeliverEventArgsValueProvider testValueProvider = new BasicDeliverEventArgsValueProvider(exceptedObject, typeof(byte[]));
 
             byte[] actualResult = (byte[])await testValueProvider.GetValueAsync();
@@ -80,6 +78,20 @@ public async Task ByteArray_Conversion_Succeeds()
             Assert.Equal(typeof(byte[]), testValueProvider.Type);
         }
 
+        [Fact]
+        public async Task ReadOnlyMemory_Byte_Conversion_Succeeds()
+        {
+            string expectedString = "someString";
+            byte[] stringInBytes = Encoding.UTF8.GetBytes(expectedString);
+            BasicDeliverEventArgs exceptedObject = new BasicDeliverEventArgs("tag", 1, false, "", "queue", null, stringInBytes);
+            BasicDeliverEventArgsValueProvider testValueProvider = new BasicDeliverEventArgsValueProvider(exceptedObject, typeof(ReadOnlyMemory<byte>));
+
+            ReadOnlyMemory<byte> actualResult = (ReadOnlyMemory<byte>)await testValueProvider.GetValueAsync();
+
+            Assert.Equal(expectedString, Encoding.UTF8.GetString(actualResult.ToArray()));
+            Assert.Equal(typeof(ReadOnlyMemory<byte>), testValueProvider.Type);
+        }
+
         [Fact]
         public async Task NormalString_Conversion_Succeeds()
         {

test/WebJobs.Extensions.RabbitMQ.Tests/PocoToBytesConverterTests.cs

@@ -19,7 +19,7 @@ public void Converts_String_Correctly()
             byte[] expectedRes = Encoding.UTF8.GetBytes(res);
 
             PocoToBytesConverter<TestClass> converter = new PocoToBytesConverter<TestClass>();
-            byte[] actualRes = converter.Convert(sampleObj);
+            byte[] actualRes = converter.Convert(sampleObj).ToArray();
 
             Assert.Equal(expectedRes, actualRes);
         }

test/WebJobs.Extensions.RabbitMQ.Tests/RabbitMQAsyncCollectorTests.cs

@@ -11,6 +11,7 @@
 using Moq;
 using RabbitMQ.Client;
 using Xunit;
+using Constants = Microsoft.Azure.WebJobs.Extensions.Constants;
 
 namespace WebJobs.Extensions.RabbitMQ.Tests
 {
@@ -44,7 +45,9 @@ public async Task AddAsync_AddsMessagesToQueue()
             byte[] body = Encoding.UTF8.GetBytes("hi");
             await collector.AddAsync(body);
 
+#pragma warning disable 618
             mockBatch.Verify(m => m.Add(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<bool>(), It.IsAny<IBasicProperties>(), body), Times.Exactly(1));
+#pragma warning restore 618
         }
     }
 }

test/WebJobs.Extensions.RabbitMQ.Tests/RabbitMQClientBuilderTests.cs

@@ -10,6 +10,7 @@
 using Moq;
 using RabbitMQ.Client;
 using Xunit;
+using Constants = Microsoft.Azure.WebJobs.Extensions.Constants;
 
 namespace WebJobs.Extensions.RabbitMQ.Tests
 {

test/WebJobs.Extensions.RabbitMQ.Tests/RabbitMQListenerTests.cs

@@ -50,12 +50,12 @@ public void CreatesHeadersAndRepublishes()
 
             RabbitMQListener listener = new RabbitMQListener(_mockExecutor.Object, _mockService.Object, "blah", _mockLogger.Object, _mockDescriptor.Object, 30);
 
-            var properties = new BasicProperties();
+            var properties = Mock.Of<IBasicProperties>();
             BasicDeliverEventArgs args = new BasicDeliverEventArgs("tag", 1, false, "", "queue", properties, Encoding.UTF8.GetBytes("hello world"));
             listener.CreateHeadersAndRepublish(args);
 
             _mockModel.Verify(m => m.BasicAck(It.IsAny<ulong>(), false), Times.Exactly(1));
-            _mockModel.Verify(m => m.BasicPublish(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IBasicProperties>(), It.IsAny<byte[]>()), Times.Exactly(1));
+            _mockModel.Verify(m => m.BasicPublish(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IBasicProperties>(), It.IsAny<ReadOnlyMemory<byte>>()), Times.Exactly(1));
         }
 
         [Fact]
@@ -64,15 +64,12 @@ public void RepublishesMessages()
             _mockService.Setup(m => m.RabbitMQModel).Returns(_mockModel.Object);
             RabbitMQListener listener = new RabbitMQListener(_mockExecutor.Object, _mockService.Object, "blah", _mockLogger.Object, _mockDescriptor.Object, 30);
 
-            var properties = new BasicProperties()
-            {
-                Headers = new Dictionary<string, object>() { { "requeueCount", 1 } }
-            };
+            var properties = Mock.Of<IBasicProperties>(property => property.Headers == new Dictionary<string, object>() { { "requeueCount", 1 } });
             BasicDeliverEventArgs args = new BasicDeliverEventArgs("tag", 1, false, "", "queue", properties, Encoding.UTF8.GetBytes("hello world"));
             listener.RepublishMessages(args);
 
             _mockModel.Verify(m => m.BasicAck(It.IsAny<ulong>(), false), Times.Exactly(1));
-            _mockModel.Verify(m => m.BasicPublish(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IBasicProperties>(), It.IsAny<byte[]>()), Times.Exactly(1));
+            _mockModel.Verify(m => m.BasicPublish(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IBasicProperties>(), It.IsAny<ReadOnlyMemory<byte>>()), Times.Exactly(1));
         }
 
         [Fact]
@@ -81,10 +78,7 @@ public void RejectsStaleMessages()
             _mockService.Setup(m => m.RabbitMQModel).Returns(_mockModel.Object);
             RabbitMQListener listener = new RabbitMQListener(_mockExecutor.Object, _mockService.Object, "blah", _mockLogger.Object, _mockDescriptor.Object, 30);
 
-            var properties = new BasicProperties()
-            {
-                Headers = new Dictionary<string, object>() { { "requeueCount", 6 } }
-            };
+            var properties = Mock.Of<IBasicProperties>(property => property.Headers == new Dictionary<string, object>() { { "requeueCount", 6 } });
             BasicDeliverEventArgs args = new BasicDeliverEventArgs("tag", 1, false, "", "queue", properties, Encoding.UTF8.GetBytes("hello world"));
             listener.RepublishMessages(args);
 

test/WebJobs.Extensions.RabbitMQ.Tests/RabbitMQTriggerBindingTests.cs

@@ -22,7 +22,7 @@ public void Verify_BindingDataContract_Types()
             expectedContract.Add("Exchange", typeof(string));
             expectedContract.Add("RoutingKey", typeof(string));
             expectedContract.Add("BasicProperties", typeof(IBasicProperties));
-            expectedContract.Add("Body", typeof(byte[]));
+            expectedContract.Add("Body", typeof(ReadOnlyMemory<byte>));
 
             var actualContract = RabbitMQTriggerBinding.CreateBindingDataContract();
 
@@ -43,9 +43,10 @@ public void Verify_BindingDataContract_Values()
             data.Add("RoutingKey", "QueueName");
 
             Random rand = new Random();
-            byte[] body = new byte[10];
-            rand.NextBytes(body);
+            byte[] buffer = new byte[10];
+            rand.NextBytes(buffer);
 
+            ReadOnlyMemory<byte> body = buffer;
             data.Add("Body", body);
 
             BasicDeliverEventArgs eventArgs = new BasicDeliverEventArgs("ConsumerName", deliveryTag, false, "n/a", "QueueName", null, body);

test/WebJobs.Extensions.RabbitMQ.Tests/WebJobs.Extensions.RabbitMQ.Tests.csproj

@@ -1,22 +1,20 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>netcoreapp2.2</TargetFramework>
+    <TargetFramework>net6.0</TargetFramework>
     <LangVersion>latest</LangVersion>
     <IsPackable>false</IsPackable>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions" Version="3.0.5" />
-    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Storage" Version="3.0.10" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="2.2.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="2.2.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="2.2.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.0.1" />
-    <PackageReference Include="Moq" Version="4.12.0" />
-    <PackageReference Include="RabbitMQ.Client" Version="5.1.2" />
-    <PackageReference Include="xunit" Version="2.4.0" />
-    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.0" />
+    <PackageReference Include="Moq" Version="4.16.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="6.2.2" />
+    <PackageReference Include="xunit" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
   </ItemGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">