Update RabbitMQ version

from our dockerhub repo to the official repo

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

Author: Gsantomaggio

.ci/publish-documentation-to-github-pages.sh

@@ -21,9 +21,9 @@ function run_docker_compose
 
 if [[ $2 == 'arm' ]]
 then
-    readonly rabbitmq_image="${RABBITMQ_IMAGE:-pivotalrabbitmq/rabbitmq-arm64:main}"
+    readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq1:4-management}"
 else
-    readonly rabbitmq_image="${RABBITMQ_IMAGE:-pivotalrabbitmq/rabbitmq:main}"
+    readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq1:4-management}"
 fi
 
 if [[ ! -v GITHUB_ACTIONS ]]

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

@@ -8,12 +8,7 @@ script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
 
-if [[ $3 == 'arm' ]]
-then
-    readonly rabbitmq_image="${RABBITMQ_IMAGE:-pivotalrabbitmq/rabbitmq-arm64:main}"
-else
-    readonly rabbitmq_image="${RABBITMQ_IMAGE:-pivotalrabbitmq/rabbitmq:main}"
-fi
+readonly rabbitmq_image="${RABBITMQ_IMAGE:-rabbitmq:4-management}"
 
 
 readonly docker_name_prefix='rabbitmq-amqp-dotnet-client'

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

@@ -1,4 +1,4 @@
 {
   "erlang": "27.0.1",
-  "rabbitmq": "4.0.0-beta.6"
+  "rabbitmq": "4.0.2"
 }

.ci/windows/versions.json

@@ -34,7 +34,7 @@ jobs:
         id: install-start-rabbitmq
         run: ${{ github.workspace }}\.ci\windows\gha-setup.ps1
       - name: Test
-        timeout-minutes: 20
+        timeout-minutes: 25
         run: dotnet test ${{ github.workspace }}\Build.csproj --no-restore --no-build --logger 'console;verbosity=detailed'
       - name: Check for errors in RabbitMQ logs
         run: ${{ github.workspace }}\.ci\windows\gha-log-check.ps1
@@ -64,7 +64,7 @@ jobs:
         id: start-rabbitmq
         run: ${{ github.workspace }}/.ci/ubuntu/cluster/gha-setup.sh
       - name: Test
-        timeout-minutes: 20
+        timeout-minutes: 25
         run: dotnet test ${{ github.workspace }}/Build.csproj --no-restore --no-build --logger "console;verbosity=detailed"
       - name: Check for errors in RabbitMQ logs
         run: ${{ github.workspace}}/.ci/ubuntu/cluster/gha-logs.sh check

.github/workflows/wf_build-and-test.yaml

@@ -9,35 +9,8 @@ build:
 test: build
 	dotnet test -c Debug $(CURDIR)/Tests/Tests.csproj --no-build --logger:"console;verbosity=detailed"
 
-rabbitmq-server-start-arm:
-	 ./.ci/ubuntu/one-node/gha-setup.sh start pull arm
+rabbitmq-server-start:
+	 ./.ci/ubuntu/one-node/gha-setup.sh start pull
 
 rabbitmq-server-stop:
 	 ./.ci/ubuntu/one-node/gha-setup.sh stop
-
-
-# TODO:
-## publish the documentation on github pages
-## you should execute this command only on the `main` branch
-# publish-github-pages:
-# 	## Create the PDF
-# 	docker run  -it -v $(shell pwd)/docs/:/client_doc/  asciidoctor/docker-asciidoctor /bin/bash -c "cd /client_doc/asciidoc &&  asciidoctor-pdf index.adoc"
-# 	## Create the HTML
-# 	docker run  -it -v $(shell pwd)/docs/:/client_doc/  asciidoctor/docker-asciidoctor /bin/bash -c "cd /client_doc/asciidoc &&  asciidoctor index.adoc"
-# 	## copy the PDF and HTML to temp folder
-# 	rm -rf docs/temp
-# 	mkdir -p docs/temp
-# 	cp docs/asciidoc/index.pdf docs/temp/dotnet-stream-client.pdf
-# 	cp docs/asciidoc/index.html docs/temp/index.html
-# 	## check out the gh-pages branch
-# 	git checkout gh-pages
-# 	## copy the PDF and HTML to the root folder
-# 	mv docs/temp/dotnet-stream-client.pdf stable/dotnet-stream-client.pdf
-# 	mv docs/temp/index.html stable/htmlsingle/index.html
-# 	## commit and push
-# 	git add stable/dotnet-stream-client.pdf
-# 	git add stable/htmlsingle/index.html
-# 	git commit -m "Update the documentation"
-# 	git push origin gh-pages
-# 	## go back to the main branch
-# 	git checkout main

Makefile

@@ -26,11 +26,63 @@ public interface IConsumer : ILifeCycle
 
     public interface IContext
     {
+        ///<summary>
+        /// Accept the message (AMQP 1.0 <code>accepted</code> outcome).
+        ///
+        ///This means the message has been processed and the broker can delete it.
+        /// 
+        /// </summary>
         Task AcceptAsync();
+
+        ///<summary>
+        /// Discard the message (AMQP 1.0 <code>rejected</code> outcome).
+        ///This means the message cannot be processed because it is invalid, the broker can drop it
+        /// or dead-letter it if it is configured.
+        ///</summary>
         Task DiscardAsync();
+
+        ///<summary>
+        ///Discard the message with annotations to combine with the existing message annotations.
+        ///This means the message cannot be processed because it is invalid, the broker can drop it
+        ///or dead-letter it if it is configured.
+        ///Application-specific annotation keys must start with the <code>x-opt-</code> prefix.
+        ///Annotation keys the broker understands starts with <code>x-</code>, but not with <code>x-opt-
+        ///</code>.
+        ///This maps to the AMQP 1.0 <code>
+        ///modified{delivery-failed = true, undeliverable-here = true}</code> outcome.
+        ///@param annotations message annotations to combine with existing ones
+        ///@see <a
+        ///    href="https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-modified">AMQP
+        ///    1.0 <code>modified</code> outcome</a>
+        ///</summary>
         Task DiscardAsync(Dictionary<string, object> annotations);
 
+        ///<summary>
+        ///Requeue the message (AMQP 1.0 <code>released</code> outcome).
+        ///
+        ///This means the message has not been processed and the broker can requeue it and deliver it
+        /// to the same or a different consumer.
+        ///
+        /// </summary>
         Task RequeueAsync();
+
+        ///<summary>
+        ///Requeue the message with annotations to combine with the existing message annotations.
+        /// 
+        ///This means the message has not been processed and the broker can requeue it and deliver it
+        /// to the same or a different consumer.
+        /// Application-specific annotation keys must start with the <code>x-opt-</code> prefix.
+        /// Annotation keys the broker understands starts with <code>x-</code>, but not with <code>x-opt-
+        /// </code>.
+        ///
+        /// This maps to the AMQP 1.0 <code>
+        /// modified{delivery-failed = false, undeliverable-here = false}</code> outcome.
+        ///
+        /// @param annotations message annotations to combine with existing ones
+        ///@see <a
+        ///     href="https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-modified">AMQP
+        ///     1.0 <code>modified</code> outcome</a>
+        ///</summary>
         Task RequeueAsync(Dictionary<string, object> annotations);
     }
 }

RabbitMQ.AMQP.Client/IConsumer.cs

@@ -62,7 +62,6 @@ public Task DiscardAsync(Dictionary<string, object> annotations)
             {
                 throw new ConsumerException("Link is closed");
             }
-
             Utils.ValidateMessageAnnotations(annotations);
 
             Task rejectTask = Task.Run(() =>

RabbitMQ.AMQP.Client/Impl/DeliveryContext.cs

@@ -24,10 +24,67 @@ public void ValidateAnnotations()
 
         const string correctAnnotationKey = "x-otp-annotation-key";
         // This should not throw an exception because the annotation key starts with "x-"
-        Utils.ValidateMessageAnnotations(new Dictionary<string, object>
-        {
-            { correctAnnotationKey, annotationValue }
-        });
+        Utils.ValidateMessageAnnotations(new Dictionary<string, object> { { correctAnnotationKey, annotationValue } });
+    }
+
+
+    [Fact]
+    public async Task RequeuedMessageWithAnnotationShouldContainAnnotationsOnRedelivery()
+    {
+        Assert.NotNull(_connection);
+        Assert.NotNull(_management);
+        TaskCompletionSource<bool> tcsRequeue =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        const string annotationKey = "x-opt-annotation-key";
+        const string annotationValue = "annotation-value";
+
+        const string annotationKey1 = "x-opt-annotation1-key";
+        const string annotationValue1 = "annotation1-value";
+
+        int requeueCount = 0;
+
+
+        await _management.Queue().Type(QueueType.QUORUM).Name(_queueName).DeclareAsync();
+        List<IMessage> messages = [];
+        IPublisher publisher = await _connection.PublisherBuilder().Queue(_queueName).BuildAsync();
+        IConsumer consumer = await _connection.ConsumerBuilder().MessageHandler(
+            async (context, message) =>
+            {
+                messages.Add(message);
+                if (requeueCount == 0)
+                {
+                    requeueCount++;
+                    await context.RequeueAsync(new Dictionary<string, object>
+                    {
+                        { annotationKey, annotationValue }, { annotationKey1, annotationValue1 }
+                    });
+                }
+                else
+                {
+                    await context.AcceptAsync();
+                    tcsRequeue.SetResult(true);
+                }
+            }
+        ).Queue(_queueName).BuildAndStartAsync();
+
+        IMessage message = new AmqpMessage($"message");
+        PublishResult pr = await publisher.PublishAsync(message);
+
+        Assert.Equal(OutcomeState.Accepted, pr.Outcome.State);
+
+        await tcsRequeue.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        Assert.Equal(2, messages.Count);
+        Assert.Null(messages[0].Annotation(annotationKey));
+        Assert.Null(messages[0].Annotation(annotationKey1));
+        Assert.Null(messages[0].Annotation("x-delivery-count"));
+
+        Assert.Equal(messages[1].Annotation(annotationKey), annotationValue);
+        Assert.Equal(messages[1].Annotation(annotationKey1), annotationValue1);
+        Assert.NotNull(messages[1].Annotation("x-delivery-count"));
+
+        await consumer.CloseAsync();
     }
 
     [Fact]
@@ -67,7 +124,6 @@ public async Task DiscardedMessageWithAnnotationsShouldBeDeadLeadLetteredAndCont
         }).Queue(dlqQueueName).BuildAndStartAsync();
 
         IMessage mResult = await tcsDl.Task.WaitAsync(TimeSpan.FromSeconds(5));
-
         Assert.NotNull(mResult);
         Assert.Equal(mResult.Annotation(annotationKey), annotationValue);
         await dlConsumer.CloseAsync();