Test for parallel batching of messages

JasperFx · Jan 15, 2024 · 47dcb38 · 47dcb38
1 parent c9e29dd
commit 47dcb38
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 3 deletions.
diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/Samples/Bootstrapping.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/Samples/Bootstrapping.cs
@@ -147,7 +147,11 @@ public async Task publishing()
                 opts.UseAmazonSqsTransport();
 
                 opts.PublishMessage<Message1>()
-                    .ToSqsQueue("outbound1");
+                    .ToSqsQueue("outbound1")
+
+                    // Increase the outgoing message throughput, but at the cost
+                    // of strict ordering 
+                    .MessageBatchMaxDegreeOfParallelism(Environment.ProcessorCount);
 
 
                 opts.PublishMessage<Message2>()

diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/send_and_receive.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/send_and_receive.cs
@@ -19,7 +19,7 @@ public async Task InitializeAsync()
 
                 opts.ListenToSqsQueue("send_and_receive");
 
-                opts.PublishAllMessages().ToSqsQueue("send_and_receive");
+                opts.PublishAllMessages().ToSqsQueue("send_and_receive").MessageBatchMaxDegreeOfParallelism(5);
             }).StartAsync();
     }
 
@@ -41,6 +41,24 @@ public async Task send_and_receive_a_single_message()
         session.Received.SingleMessage<SqsMessage>()
             .Name.ShouldBe(message.Name);
     }
+
+    [Fact]
+    public async Task send_and_receive_many_messages()
+    {
+        Func<IMessageBus, Task> sending = async bus =>
+        {
+            for (int i = 0; i < 100; i++)
+            {
+                await bus.PublishAsync(new SqsMessage(Guid.NewGuid().ToString()));
+            }
+        };
+
+        await _host.TrackActivity()
+            .IncludeExternalTransports()
+            .Timeout(5.Minutes())
+            .ExecuteAndWaitAsync(sending);
+
+    }
 }
 
 public record SqsMessage(string Name);

diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs
@@ -27,6 +27,8 @@ internal AmazonSqsQueue(string queueName, AmazonSqsTransport parent) : base(new
         EndpointName = queueName;
 
         Configuration = new CreateQueueRequest(QueueName);
+
+        MessageBatchSize = 10;
     }
 
     /// <summary>

diff --git a/src/Wolverine/Configuration/SubscriberConfiguration.cs b/src/Wolverine/Configuration/SubscriberConfiguration.cs
@@ -71,7 +71,8 @@ public T MessageBatchSize(int batchSize)
 
     /// <summary>
     /// For endpoints that send messages in batches, this governs the maximum number
-    /// of concurrent outgoing batches 
+    /// of concurrent outgoing batches. The default is 1 to ensure message order, but increase this
+    /// number to improve outgoing throughput 
     /// </summary>
     public T MessageBatchMaxDegreeOfParallelism(int batchMaxDegreeOfParallelism)
     {