[Host.AzureServiceBus][Host.AzureEventHub] Fix the batch publish chunking iterations

Signed-off-by: Tomasz Maruszak <[email protected]>

Author: zarusz

src/Host.Plugin.Properties.xml

@@ -4,7 +4,7 @@
   <Import Project="Common.NuGet.Properties.xml" />
 
   <PropertyGroup>
-    <Version>3.3.1-rc103</Version>
+    <Version>3.3.1</Version>
   </PropertyGroup>
 
 </Project>

src/SlimMessageBus.Host.AzureEventHub/EventHubMessageBus.cs

@@ -187,21 +187,31 @@ public override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBu
                         {
                             inBatch.Add(item.Envelope);
                             advance = it.MoveNext();
-                            if (advance)
+                        }
+                        else
+                        {
+                            // Current message doesn't fit in this batch
+                            if (batch.Count == 0)
                             {
-                                continue;
+                                throw new ProducerMessageBusException($"Failed to add message {item.Envelope.Message} of type {item.Envelope.MessageType?.Name} on path {path} to an empty batch");
                             }
-                        }
 
-                        if (batch.Count == 0)
-                        {
-                            throw new ProducerMessageBusException($"Failed to add message {item.Envelope.Message} of type {item.Envelope.MessageType?.Name} on path {path} to an empty batch");
+                            // Send the current batch and retry with the current message in a new batch
+                            await producer.SendAsync(batch, cancellationToken).ConfigureAwait(false);
+                            dispatched.AddRange(inBatch);
+                            inBatch.Clear();
+
+                            batch.Dispose();
+                            batch = null;
+                            // Don't advance - retry the current message in the next iteration
                         }
+                    }
 
-                        advance = false;
+                    // Send any remaining messages in the final batch
+                    if (batch != null && batch.Count > 0)
+                    {
                         await producer.SendAsync(batch, cancellationToken).ConfigureAwait(false);
                         dispatched.AddRange(inBatch);
-                        inBatch.Clear();
 
                         batch.Dispose();
                         batch = null;

src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs

@@ -251,22 +251,32 @@ public override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBu
                 {
                     inBatch.Add(item.Envelope);
                     advance = it.MoveNext();
-                    if (advance)
+                }
+                else
+                {
+                    // Current message doesn't fit in this batch
+                    if (batch.Count == 0)
                     {
-                        continue;
+                        throw new ProducerMessageBusException($"Failed to add message {item.Envelope.Message} of Type {item.Envelope.MessageType?.Name} on Path {path} to an empty batch");
                     }
-                }
 
-                if (batch.Count == 0)
-                {
-                    throw new ProducerMessageBusException($"Failed to add message {item.Envelope.Message} of Type {item.Envelope.MessageType?.Name} on Path {path} to an empty batch");
+                    // Send the current batch and retry with the current message in a new batch
+                    await SendBatchAsync(path, senderClient, envelopes, batch, cancellationToken).ConfigureAwait(false);
+                    dispatched.AddRange(inBatch);
+                    inBatch.Clear();
+
+                    batch.Dispose();
+                    batch = null;
+                    // Don't advance - retry the current message in the next iteration
                 }
+            }
 
-                advance = false;
+            // Send any remaining messages in the final batch
+            if (batch != null && batch.Count > 0)
+            {
                 await SendBatchAsync(path, senderClient, envelopes, batch, cancellationToken).ConfigureAwait(false);
                 dispatched.AddRange(inBatch);
-                inBatch.Clear();
-
+                
                 batch.Dispose();
                 batch = null;
             }

src/SlimMessageBus.Host.Configuration/SlimMessageBus.Host.Configuration.csproj

@@ -6,7 +6,7 @@
     <Description>Core configuration interfaces of SlimMessageBus</Description>
     <PackageTags>SlimMessageBus</PackageTags>
     <RootNamespace>SlimMessageBus.Host</RootNamespace>
-    <Version>3.3.1-rc103</Version>
+    <Version>3.3.1</Version>
   </PropertyGroup>
 
   <ItemGroup>