Refactors

Author: lukebakken

Tests/Consumer/ConsumerOutcomeTests.cs

@@ -170,7 +170,6 @@ public async Task DiscardedMessageShouldBeDeadLeadLetteredWhenConfigured()
         string dlqQueueName = $"dlq_{_queueName}";
         await DeclareDeadLetterTopology(_queueName, dlqQueueName);
 
-        //this.connection.consumerBuilder().queue(q).messageHandler((ctx, msg) -> ctx.discard()).build();
         IConsumerBuilder queueConsumerBuilder = _connection.ConsumerBuilder()
             .Queue(_queueName)
             .MessageHandler((cxt, msg) =>

Tests/Consumer/ConsumerPauseTests.cs

@@ -220,16 +220,7 @@ public async Task ConsumerGracefulShutdownExample()
 
         consumer.Pause();
 
-        DateTime start = DateTime.Now;
-        while (consumer.UnsettledMessageCount != 0)
-        {
-            await Task.Delay(TimeSpan.FromMilliseconds(100));
-            DateTime now = DateTime.Now;
-            if (start - now > _waitSpan)
-            {
-                Assert.Fail("consumer.UnsettledMessageCount never reached zero!");
-            }
-        }
+        await WaitUntilStable(() => consumer.UnsettledMessageCount, 0);
 
         await consumer.CloseAsync();
         consumer.Dispose();

Tests/Consumer/StreamConsumerTests.cs

@@ -506,7 +506,7 @@ Task MessageHandler(IContext cxt, IMessage msg)
 
         using (IConsumer consumer = await consumerBuilder0.BuildAndStartAsync())
         {
-            await WaitUntilStable(() => (int)Interlocked.Read(ref receivedCount));
+            await WaitUntilStable<long>(() => Interlocked.Read(ref receivedCount));
             await consumer.CloseAsync();
         }
 

Tests/IntegrationTest.cs

@@ -171,49 +171,51 @@ protected Task WhenTaskCompletes(Task task)
         return task.WaitAsync(_waitSpan);
     }
 
-    protected async Task WaitUntilStable(Func<int> getValue, int? stableValue = null)
+    protected static async Task WaitUntilStable<T>(Func<T> getValue, T? stableValue = default)
+        where T : IEquatable<T>
     {
         const int iterations = 20;
         int iteration = 0;
-        int val0 = int.MinValue;
-        int val1 = int.MaxValue;
+        T val0;
+        T val1;
         do
         {
             await Task.Delay(250);
             val0 = getValue();
             await Task.Delay(250);
             val1 = getValue();
             iteration++;
-            if (stableValue is not null && val1 == stableValue)
+            if (stableValue is not null && val1.Equals(stableValue))
             {
                 break;
             }
-        } while (val0 != val1 && iteration < iterations);
+        } while (false == val0.Equals(val1) && iteration < iterations);
 
         if (iteration == iterations)
         {
             Assert.Fail($"value did not stabilize as expected (stableValue: {stableValue})");
         }
     }
 
-    protected async Task WaitUntilStable(Func<Task<int>> getValue, int? stableValue = null)
+    protected static async Task WaitUntilStable<T>(Func<Task<T>> getValue, T? stableValue = default)
+        where T : IEquatable<T>
     {
         const int iterations = 20;
         int iteration = 0;
-        int val0 = int.MinValue;
-        int val1 = int.MaxValue;
+        T val0;
+        T val1;
         do
         {
             await Task.Delay(250);
             val0 = await getValue();
             await Task.Delay(250);
             val1 = await getValue();
             iteration++;
-            if (stableValue is not null && val1 == stableValue)
+            if (stableValue is not null && val1.Equals(stableValue))
             {
                 break;
             }
-        } while (val0 != val1 && iteration < iterations);
+        } while (false == val0.Equals(val1) && iteration < iterations);
 
         if (iteration == iterations)
         {