add tests

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

RabbitMQ.AMQP.Client/Impl/AmqpMessage.cs

@@ -24,14 +24,20 @@ public AmqpMessage()
             NativeMessage = new Message();
         }
 
+        /// <summary>
+        /// Create a message with a body of type byte[] and BodySection of type Data.
+        /// </summary>
+        /// <param name="body"></param>
         public AmqpMessage(byte[] body)
         {
             NativeMessage = new Message();
             NativeMessage.BodySection = new Data { Binary = body };
         }
 
-        // This constructor is used for string body
-        // Like AmqpMessage(byte[] body) but for string with UTF8 encoding
+        /// <summary>
+        /// Create a message with a body of type string and BodySection of type Data.
+        /// The string is converted to a byte[] using UTF8 encoding.
+        /// </summary>
         public AmqpMessage(string body)
         {
             NativeMessage = new Message();

Tests/Amqp091/FromToAmqp091Tests.cs

@@ -19,15 +19,14 @@ public async Task ToAmqp091()
         {
             Assert.NotNull(_connection);
             Assert.NotNull(_management);
-            _queueName = "q";
 
             IQueueSpecification queueSpec = _management.Queue().Name(_queueName).Type(QueueType.QUORUM);
             await queueSpec.DeclareAsync();
 
             var publisher = await _connection.PublisherBuilder().BuildAsync();
-            byte[] body = System.Text.Encoding.UTF8.GetBytes("{Text:as,Seq:1,Max:7000}");
+            const string body = "{Text:as,Seq:1,Max:7000}";
             IMessage amqpMessage = new AmqpMessage(body).ToAddress().Queue(_queueName).Build();
-            for (int i = 0; i < 1; i++)
+            for (int i = 0; i < 0; i++)
             {
                 PublishResult result = await publisher.PublishAsync(message: amqpMessage).ConfigureAwait(true);
                 Assert.NotNull(result);
@@ -51,6 +50,45 @@ public async Task ToAmqp091()
             Assert.Equal("consumerTag", receivedMessage091.ConsumerTag);
             Assert.Equal("{Text:as,Seq:1,Max:7000}",
                 System.Text.Encoding.UTF8.GetString(receivedMessage091.Body.ToArray()));
+            channel.Close();
+            connection.Close();
+        }
+
+        [Fact]
+        public async Task FromAmqp091()
+        {
+            Assert.NotNull(_connection);
+            Assert.NotNull(_management);
+
+            IQueueSpecification queueSpec = _management.Queue().Name(_queueName).Type(QueueType.QUORUM);
+            await queueSpec.DeclareAsync();
+
+            // publish a message with AMQP 0-9-1
+            var factory = new ConnectionFactory();
+            var connection = factory.CreateConnection();
+            var channel = connection.CreateModel();
+            channel.BasicPublish(
+                exchange: "",
+                routingKey: _queueName,
+                basicProperties: null,
+                body: System.Text.Encoding.UTF8.GetBytes("{Text:as,Seq:1,Max:7000}"));
+
+            TaskCompletionSource<IMessage> tcs = new();
+            IConsumer consumer = await _connection.ConsumerBuilder()
+                .Queue(_queueName)
+                .MessageHandler((context, message) =>
+                {
+                    tcs.SetResult(message);
+                    context.Accept();
+                    return Task.CompletedTask;
+                }).BuildAndStartAsync();
+
+            var receivedMessage = await tcs.Task;
+            Assert.NotNull(receivedMessage);
+            Assert.Equal("{Text:as,Seq:1,Max:7000}", receivedMessage.BodyAsString());
+            channel.Close();
+            connection.Close();
+            await consumer.CloseAsync();
         }
     }
 }