Merge pull request #97 from rabbitmq/lukebakken/test-flakes

Address GHA test flakes

Author: lukebakken

.ci/windows/versions.json

@@ -1,4 +1,4 @@
 {
   "erlang": "27.1.2",
-  "rabbitmq": "4.0.3"
+  "rabbitmq": "4.0.4"
 }

RabbitMQ.AMQP.Client/Extensions.cs

@@ -56,7 +56,7 @@ public static Task<T> WaitAsync<T>(this Task<T> task, CancellationToken cancella
 
         private static async Task DoWaitAsync(this Task task, CancellationToken cancellationToken)
         {
-            var cancellationTokenTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            TaskCompletionSource<bool> cancellationTokenTcs = Utils.CreateTaskCompletionSource<bool>();
 
             using (cancellationToken.Register(s => ((TaskCompletionSource<bool>)s).TrySetResult(true),
                 state: cancellationTokenTcs, useSynchronizationContext: false))
@@ -73,7 +73,7 @@ private static async Task DoWaitAsync(this Task task, CancellationToken cancella
 
         private static async Task<T0> DoWaitGenericAsync<T0>(this Task<T0> task, CancellationToken cancellationToken)
         {
-            var cancellationTokenTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            TaskCompletionSource<bool> cancellationTokenTcs = Utils.CreateTaskCompletionSource<bool>();
 
             using (cancellationToken.Register(s => ((TaskCompletionSource<bool>)s).TrySetResult(true),
                 state: cancellationTokenTcs, useSynchronizationContext: false))

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -96,7 +96,7 @@ internal void RemoveConsumer(Guid id)
         }
 
         private readonly TaskCompletionSource<bool> _connectionClosedTcs =
-            new(TaskCreationOptions.RunContinuationsAsynchronously);
+            Utils.CreateTaskCompletionSource<bool>();
 
         public IRpcServerBuilder RpcServerBuilder()
         {

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -46,7 +46,7 @@ public override async Task OpenAsync()
             try
             {
                 TaskCompletionSource<ReceiverLink> attachCompletedTcs =
-                    new(TaskCreationOptions.RunContinuationsAsynchronously);
+                    Utils.CreateTaskCompletionSource<ReceiverLink>();
 
                 // this is an event to get the filters to the listener context
                 // it _must_ be here because in case of reconnect the original filters could be not valid anymore

RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs

@@ -47,7 +47,7 @@ public class AmqpManagement : AbstractLifeCycle, IManagement, IManagementTopolog
         private const string ReplyTo = "$me";
 
         protected readonly TaskCompletionSource<bool> _managementSessionClosedTcs =
-            new(TaskCreationOptions.RunContinuationsAsynchronously);
+            Utils.CreateTaskCompletionSource<bool>();
 
         internal AmqpManagement(AmqpManagementParameters amqpManagementParameters)
         {
@@ -269,7 +269,7 @@ private async Task EnsureReceiverLinkAsync()
                         new Target() { Address = ManagementNodeAddress, ExpiryPolicy = new Symbol("SESSION_END"), },
                 };
 
-                var tcs = new TaskCompletionSource<ReceiverLink>(TaskCreationOptions.RunContinuationsAsynchronously);
+                TaskCompletionSource<ReceiverLink> tcs = Utils.CreateTaskCompletionSource<ReceiverLink>();
                 var tmpReceiverLink = new ReceiverLink(
                     _managementSession, LinkPairName, receiveAttach, (ILink link, Attach attach) =>
                     {
@@ -328,7 +328,7 @@ private async Task EnsureSenderLinkAsync()
                     },
                 };
 
-                var tcs = new TaskCompletionSource<SenderLink>(TaskCreationOptions.RunContinuationsAsynchronously);
+                TaskCompletionSource<SenderLink> tcs = Utils.CreateTaskCompletionSource<SenderLink>();
                 var tmpSenderLink = new SenderLink(
                     _managementSession, LinkPairName, senderAttach, (ILink link, Attach attach) =>
                     {
@@ -425,10 +425,10 @@ internal async Task<Message> RequestAsync(Message message, int[] expectedRespons
             // TODO: make the timeout configurable
             TimeSpan timeout = argTimeout ?? TimeSpan.FromSeconds(30);
 
-            TaskCompletionSource<Message> mre = new(TaskCreationOptions.RunContinuationsAsynchronously);
+            TaskCompletionSource<Message> tcs = Utils.CreateTaskCompletionSource<Message>();
 
             // Add TaskCompletionSource to the dictionary it will be used to set the result of the request
-            if (false == _requests.TryAdd(message.Properties.MessageId, mre))
+            if (false == _requests.TryAdd(message.Properties.MessageId, tcs))
             {
                 // TODO what to do in this error case?
             }
@@ -461,7 +461,7 @@ void RequestTimeoutAction()
 
             // The response is handled in a separate thread, see ProcessResponses method in the Init method
             // TODO timeout & token
-            Message result = await mre.Task.WaitAsync(linkedCts.Token)
+            Message result = await tcs.Task.WaitAsync(linkedCts.Token)
                 .ConfigureAwait(false);
 
             await sendTask.WaitAsync(linkedCts.Token)

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -35,7 +35,7 @@ public override async Task OpenAsync()
             try
             {
                 TaskCompletionSource<SenderLink> attachCompletedTcs =
-                    new(TaskCreationOptions.RunContinuationsAsynchronously);
+                    Utils.CreateTaskCompletionSource<SenderLink>();
 
                 Attach attach = Utils.CreateAttach(_address, DeliveryMode.AtLeastOnce, _id);
 
@@ -122,7 +122,8 @@ public async Task<PublishResult> PublishAsync(IMessage message, CancellationToke
             try
             {
                 TaskCompletionSource<PublishOutcome> messagePublishedTcs =
-                    new(TaskCreationOptions.RunContinuationsAsynchronously);
+                    Utils.CreateTaskCompletionSource<PublishOutcome>();
+
                 Message nativeMessage = ((AmqpMessage)message).NativeMessage;
 
                 void OutcomeCallback(ILink sender, Message inMessage, Outcome outcome, object state)

RabbitMQ.AMQP.Client/Impl/AmqpQueueSpecification.cs

@@ -35,7 +35,16 @@ internal DefaultQueueInfo(string queueName)
 
         internal DefaultQueueInfo(Map response)
         {
-            _name = (string)response["name"];
+            if (response["name"] is string name)
+            {
+                _name = name;
+            }
+            else
+            {
+                // TODO error?
+                _name = string.Empty;
+            }
+
             _durable = (bool)response["durable"];
             _autoDelete = (bool)response["auto_delete"];
             _exclusive = (bool)response["exclusive"];
@@ -48,10 +57,12 @@ internal DefaultQueueInfo(Map response)
 
             _leader = (string)response["leader"];
 
-            string[]? replicas = (string[])response["replicas"];
-            if (replicas.Length > 0)
+            if (response["replicas"] is string[] queueReplicas)
             {
-                _replicas.AddRange(replicas);
+                if (queueReplicas.Length > 0)
+                {
+                    _replicas.AddRange(queueReplicas);
+                }
             }
 
             _messageCount = (ulong)response["message_count"];

RabbitMQ.AMQP.Client/Impl/AmqpRpcClient.cs

@@ -158,6 +158,7 @@ public override async Task OpenAsync()
                 .Queue(_configuration.ReplyToQueue)
                 .MessageHandler((context, message) =>
                 {
+                    // TODO MessageHandler funcs should catch all exceptions
                     context.Accept();
                     object correlationId = ExtractCorrelationId(message);
                     if (_pendingRequests.TryGetValue(correlationId, out TaskCompletionSource<IMessage>? request))
@@ -198,8 +199,7 @@ public async Task<IMessage> PublishAsync(IMessage message, CancellationToken can
             {
                 object correlationId = CorrelationIdSupplier();
                 message = RequestPostProcess(message, correlationId);
-                _pendingRequests.TryAdd(correlationId,
-                    new TaskCompletionSource<IMessage>(TaskCreationOptions.RunContinuationsAsynchronously));
+                _pendingRequests.TryAdd(correlationId, Utils.CreateTaskCompletionSource<IMessage>());
                 if (_publisher != null)
                 {
                     PublishResult pr = await _publisher.PublishAsync(

RabbitMQ.AMQP.Client/Impl/AmqpSessionManagement.cs

@@ -34,7 +34,7 @@ internal async Task<Session> GetOrCreateSessionAsync()
             }
             else
             {
-                TaskCompletionSource<ISession> sessionBeginTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+                TaskCompletionSource<ISession> sessionBeginTcs = Utils.CreateTaskCompletionSource<ISession>();
                 void OnBegin(ISession session, Begin peerBegin)
                 {
                     sessionBeginTcs.SetResult(session);

RabbitMQ.AMQP.Client/Utils.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Runtime.CompilerServices;
 using System.Security.Cryptography;
 using System.Text;
 using System.Threading.Tasks;
@@ -39,6 +40,12 @@ internal static int RandomNext(int minValue = 0, int maxValue = 1024)
 #endif
         }
 
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal static TaskCompletionSource<T> CreateTaskCompletionSource<T>()
+        {
+            return new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
+        }
+
         internal static string GenerateQueueName()
         {
             return GenerateName(DefaultPrefix);