Use RabbitMQ BasicProperties (#156)

Author: marinasundstrom

src/MyServiceBus.RabbitMq/RabbitMqQueueSendTransport.cs

@@ -22,8 +22,10 @@ public async Task Send<T>(T message, SendContext context, CancellationToken canc
     {
         var body = await context.Serialize(message);
 
-        var props = _channel.CreateBasicProperties();
-        props.Persistent = true;
+        var props = new BasicProperties
+        {
+            Persistent = true
+        };
 
         if (context.Headers != null)
         {

src/MyServiceBus.RabbitMq/RabbitMqSendTransport.cs

@@ -18,8 +18,10 @@ public async Task Send<T>(T message, SendContext context, CancellationToken canc
     {
         var body = await context.Serialize(message);
 
-        var props = _channel.CreateBasicProperties();
-        props.Persistent = true;
+        var props = new BasicProperties
+        {
+            Persistent = true
+        };
 
         if (context.Headers != null)
         {

src/MyServiceBus.RabbitMq/RabbitMqTransportFactory.cs

@@ -18,7 +18,7 @@ public RabbitMqTransportFactory(ConnectionProvider connectionProvider)
         _connectionProvider = connectionProvider;
     }
 
-    [Throws(typeof(OverflowException), typeof(InvalidOperationException), typeof(ArgumentException))]
+    [Throws(typeof(OverflowException), typeof(InvalidOperationException), typeof(ArgumentException), typeof(OperationCanceledException))]
     public async Task<ISendTransport> GetSendTransport(Uri address, CancellationToken cancellationToken = default)
     {
         string? exchange = null;
@@ -133,6 +133,7 @@ await channel.QueueDeclareAsync(
         return sendTransport;
     }
 
+    [Throws(typeof(ObjectDisposedException))]
     public async Task<IReceiveTransport> CreateReceiveTransport(
         ReceiveEndpointTopology topology,
         Func<ReceiveContext, Task> handler,
@@ -150,6 +151,7 @@ await channel.ExchangeDeclareAsync(
         );
 
         var hasErrorQueue = !topology.AutoDelete;
+        IDictionary<string, object?>? mainQueueArguments = null;
 
         if (hasErrorQueue)
         {
@@ -178,13 +180,19 @@ await channel.QueueBindAsync(
                 routingKey: string.Empty,
                 cancellationToken: cancellationToken
             );
+
+            mainQueueArguments = new Dictionary<string, object?>
+            {
+                ["x-dead-letter-exchange"] = errorExchange
+            };
         }
 
         await channel.QueueDeclareAsync(
             queue: topology.QueueName,
             durable: topology.Durable,
             exclusive: false,
             autoDelete: topology.AutoDelete,
+            arguments: mainQueueArguments,
             cancellationToken: cancellationToken
         );
 

test/MyServiceBus.RabbitMq.Tests/HeaderEncodingTests.cs

@@ -8,6 +8,8 @@
 using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
+using MyServiceBus;
+using MyServiceBus.Serialization;
 
 namespace MyServiceBus.RabbitMq.Tests;
 
@@ -22,21 +24,20 @@ class FaultingConsumer : IConsumer<TestMessage>
     }
 
     [Fact]
-    [Throws(typeof(UriFormatException), typeof(ArgumentException), typeof(InvalidOperationException))]
+    [Throws(typeof(UriFormatException), typeof(ArgumentException), typeof(InvalidOperationException), typeof(JsonException))]
     public async Task Faulted_message_headers_include_mt_prefix()
     {
         var channel = Substitute.For<IChannel>();
-        IBasicProperties? captured = null;
-        channel.CreateBasicProperties().Returns(new BasicProperties());
+        BasicProperties? captured = null;
         channel.BasicPublishAsync(
             Arg.Any<string>(),
             Arg.Any<string>(),
             Arg.Any<bool>(),
-            Arg.Any<IBasicProperties>(),
+            Arg.Any<BasicProperties>(),
             Arg.Any<ReadOnlyMemory<byte>>(),
             Arg.Any<CancellationToken>())
-            .Returns(Task.CompletedTask)
-            .AndDoes(ci => captured = ci.Arg<IBasicProperties>());
+            .Returns(_ => ValueTask.CompletedTask)
+            .AndDoes(ci => captured = ci.Arg<BasicProperties>());
 
         var services = new ServiceCollection();
         services.AddTransient<FaultingConsumer>();

test/MyServiceBus.RabbitMq.Tests/RabbitMqTransportFactoryTests.cs

@@ -13,7 +13,7 @@ namespace MyServiceBus.RabbitMq.Tests;
 
 public class RabbitMqTransportFactoryTests
 {
-    [Fact(Skip = "Not working")]
+    [Fact]
     [Throws(typeof(Exception))]
     public async Task Declares_dead_letter_exchange_and_queue()
     {
@@ -164,7 +164,7 @@ await channel.DidNotReceive().QueueDeclareAsync(
     }
 
     [Fact]
-    [Throws(typeof(OverflowException))]
+    [Throws(typeof(OverflowException), typeof(InvalidOperationException))]
     public async Task Supports_exchange_scheme_uri()
     {
         var channel = Substitute.For<IChannel>();
@@ -205,7 +205,7 @@ await channel.Received(1).ExchangeDeclareAsync(
     }
 
     [Fact]
-    [Throws(typeof(OverflowException))]
+    [Throws(typeof(OverflowException), typeof(InvalidOperationException))]
     public async Task Supports_queue_scheme_uri()
     {
         var channel = Substitute.For<IChannel>();

test/MyServiceBus.RabbitMq.Tests/TransportHeaderTests.cs

@@ -5,6 +5,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Xunit;
+using MyServiceBus;
 using MyServiceBus.Serialization;
 
 namespace MyServiceBus.RabbitMq.Tests;
@@ -14,20 +15,20 @@ public class TransportHeaderTests
     class TestMessage { }
 
     [Fact]
+    [Throws(typeof(NotSupportedException))]
     public async Task Underscore_headers_are_applied_to_basic_properties()
     {
         var channel = Substitute.For<IChannel>();
-        IBasicProperties? captured = null;
-        channel.CreateBasicProperties().Returns(new BasicProperties());
+        BasicProperties? captured = null;
         channel.BasicPublishAsync(
             Arg.Any<string>(),
             Arg.Any<string>(),
             Arg.Any<bool>(),
-            Arg.Any<IBasicProperties>(),
+            Arg.Any<BasicProperties>(),
             Arg.Any<ReadOnlyMemory<byte>>(),
             Arg.Any<CancellationToken>())
-            .Returns(Task.CompletedTask)
-            .AndDoes(ci => captured = ci.Arg<IBasicProperties>());
+            .Returns(_ => ValueTask.CompletedTask)
+            .AndDoes(ci => captured = ci.Arg<BasicProperties>());
 
         var transport = new RabbitMqSendTransport(channel, "test");
         var context = new SendContext(MessageTypeCache.GetMessageTypes(typeof(TestMessage)), new EnvelopeMessageSerializer())