consume form 091

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

Author: Gsantomaggio

.ci/ubuntu/cluster/gha-setup.sh

@@ -19,7 +19,7 @@ function run_docker_compose
     docker compose --file "$script_dir/docker-compose.yml" $@
 }
 
-readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq:4.1.0-beta.4-management-alpine}"
+readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq:4.1.0-management-alpine}"
 
 if [[ ! -v GITHUB_ACTIONS ]]
 then

.ci/ubuntu/one-node/gha-setup.sh

@@ -9,7 +9,7 @@ readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
 
 
-readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq:4.1.0-beta.4-management-alpine}"
+readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq:4.1.0-management-alpine}"
 
 
 

.ci/windows/versions.json

@@ -1,4 +1,4 @@
 {
   "erlang": "27.2",
-  "rabbitmq": "4.1.0-beta.4"
+  "rabbitmq": "4.1.0"
 }

.gitignore

@@ -129,4 +129,5 @@ docs/temp/
 
 # ci logs
 .ci/ubuntu/log/*
+.ci/ubuntu/cluster/log/*
 

Directory.Packages.props

@@ -7,6 +7,7 @@
     <PackageVersion Include="AMQPNetLite.Core" Version="2.4.11" />
     <PackageVersion Include="OpenTelemetry" Version="1.10.0" />
     <PackageVersion Include="OpenTelemetry.Exporter.Console" Version="1.10.0" />
+    <PackageVersion Include="RabbitMQ.Client" Version="7.1.2" />
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="9.0.0" />
     <!-- HAClient -->
     <PackageVersion Include="DotNext.Threading" Version="5.15.0" />

Tests/Amqp091/FromToAmqp091Tests.cs

@@ -0,0 +1,80 @@
+// This source code is dual-licensed under the Apache License, version 2.0,
+// and the Mozilla Public License, version 2.0.
+// Copyright (c) 2017-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Newtonsoft.Json;
+using RabbitMQ.AMQP.Client;
+using RabbitMQ.AMQP.Client.Impl;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Tests.Amqp091
+{
+    public class FromToAmqp091Tests(ITestOutputHelper testOutputHelper) : IntegrationTest(testOutputHelper)
+    {
+        [Fact]
+        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();
+            // string text = JsonConvert.SerializeObject(message); //produces {"Text":"as","Seq":1,"Max":7000} 
+            byte[] body = System.Text.Encoding.UTF8.GetBytes("{Text:as,Seq:1,Max:7000}");
+            IMessage amqpMessage = new AmqpMessage(body).ToAddress().Queue(_queueName).Build();
+            for (int i = 0; i < 100; i++)
+            {
+                PublishResult result = await publisher.PublishAsync(message: amqpMessage).ConfigureAwait(true);
+                Assert.NotNull(result);
+                Assert.Equal(OutcomeState.Accepted, result.Outcome.State);
+            }
+
+
+            // TaskCompletionSource<IMessage> tcs = new();
+            // IConsumer consumer = await _connection.ConsumerBuilder()
+            //     .Queue(queueSpec)
+            //     .MessageHandler((context, message) =>
+            //         {
+            //             tcs.SetResult(message);
+            //             context.Accept();
+            //             return Task.CompletedTask;
+            //         }
+            //     ).BuildAndStartAsync();
+            //
+            // IMessage receivedMessage = await tcs.Task;
+            // Assert.NotNull(receivedMessage);
+            // // get the string form bytes
+            //
+            // string receivedMessageBody = System.Text.Encoding.UTF8.GetString((byte[])receivedMessage.Body());
+            // Assert.Equal("{Text:as,Seq:1,Max:7000}", receivedMessageBody);
+
+
+            var factory = new ConnectionFactory();
+            var connection = await factory.CreateConnectionAsync();
+            var channel = await connection.CreateChannelAsync();
+            var consumer091 = new AsyncEventingBasicConsumer(channel);
+            var tcs091 = new TaskCompletionSource<BasicDeliverEventArgs>();
+            consumer091.ReceivedAsync += async (sender, ea) =>
+            {
+                tcs091.SetResult(ea);
+                await channel.BasicAckAsync(deliveryTag: ea.DeliveryTag, multiple: false);
+            };
+            await channel.BasicConsumeAsync(_queueName, false, "consumerTag", consumer091);
+            var receivedMessage091 = await tcs091.Task;
+            Assert.NotNull(receivedMessage091);
+            Assert.Equal(_queueName, receivedMessage091.RoutingKey);
+            Assert.Equal("consumerTag", receivedMessage091.ConsumerTag);
+            Assert.Equal("{Text:as,Seq:1,Max:7000}",
+                System.Text.Encoding.UTF8.GetString(receivedMessage091.Body.ToArray()));
+        }
+    }
+}

Tests/Tests.csproj

@@ -29,6 +29,7 @@
     <PackageReference Include="Microsoft.Extensions.Diagnostics.Testing" />
     <PackageReference Include="Microsoft.Extensions.Diagnostics" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" />
+    <PackageReference Include="RabbitMQ.Client" />
     <PackageReference Include="System.IdentityModel.Tokens.Jwt" />
     <PackageReference Include="System.Text.Json" />
     <PackageReference Include="xunit" />