Fix flaky test

Sometimes waiting 2 seconds is not long enough:

https://ci.rabbitmq.com/teams/main/pipelines/dotnet/jobs/test/builds/240

refactor

Fix WorkPool to execute all jobs in the Work queue in a loop iteration

Author: lukebakken

projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs

@@ -43,9 +43,8 @@ class WorkPool
             readonly ConcurrentQueue<Work> _workQueue;
             readonly CancellationTokenSource _tokenSource;
             readonly ModelBase _model;
-            CancellationTokenRegistration _tokenRegistration;
+            readonly CancellationTokenRegistration _tokenRegistration;
             volatile TaskCompletionSource<bool> _syncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-            private Task _task;
 
             public WorkPool(ModelBase model)
             {
@@ -57,7 +56,7 @@ public WorkPool(ModelBase model)
 
             public void Start()
             {
-                _task = Task.Run(Loop, CancellationToken.None);
+                Task.Run(Loop, CancellationToken.None);
             }
 
             public void Enqueue(Work work)
@@ -80,7 +79,7 @@ async Task Loop()
                         // Swallowing the task cancellation in case we are stopping work.
                     }
 
-                    if (!_tokenSource.IsCancellationRequested && _workQueue.TryDequeue(out Work work))
+                    while (_tokenSource.IsCancellationRequested == false && _workQueue.TryDequeue(out Work work))
                     {
                         await work.Execute(_model).ConfigureAwait(false);
                     }

projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs

@@ -336,7 +336,9 @@ public void FinishClose()
         public void HandleCommand(ISession session, Command cmd)
         {
             if (!DispatchAsynchronous(cmd))// Was asynchronous. Already processed. No need to process further.
+            {
                 _continuationQueue.Next().HandleCommand(cmd);
+            }
         }
 
         public MethodBase ModelRpc(MethodBase method, ContentHeaderBase header, byte[] body)

projects/client/RabbitMQ.Client/src/client/impl/SocketFrameHandler.cs

@@ -202,7 +202,6 @@ public InboundFrame ReadFrame()
             return RabbitMQ.Client.Impl.InboundFrame.ReadFrom(_reader);
         }
 
-        private static readonly byte[] s_amqp = Encoding.ASCII.GetBytes("AMQP");
         public void SendHeader()
         {
             byte[] headerBytes = new byte[8];