Prevent dequeue empty blocked requests queue

fsprojects · Jul 23, 2023 · 11e6aa0 · 11e6aa0
1 parent 479a3e4
commit 11e6aa0
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 7 deletions.
diff --git a/paket.lock b/paket.lock
@@ -217,7 +217,7 @@ NUGET
       System.Collections.Immutable (>= 7.0)
     protobuf-net.Reflection (3.2.12)
       protobuf-net.Core (>= 3.2.12)
-    Pulsar.Client (2.12.4)
+    Pulsar.Client (2.12.6)
       Apache.Avro (>= 1.11.2)
       AvroSchemaGenerator (>= 2.9.2)
       FSharp.Core (>= 7.0.300)
@@ -369,7 +369,7 @@ NUGET
       System.Memory (>= 4.5.5)
     protobuf-net.Reflection (3.2.12)
       protobuf-net.Core (>= 3.2.12)
-    Pulsar.Client (2.12.4)
+    Pulsar.Client (2.12.6)
       Apache.Avro (>= 1.11.2)
       AvroSchemaGenerator (>= 2.9.2)
       FSharp.Core (>= 7.0.300)

diff --git a/src/Pulsar.Client/Internal/ProducerImpl.fs b/src/Pulsar.Client/Internal/ProducerImpl.fs
@@ -197,10 +197,12 @@ type internal ProducerImpl<'T> private (producerConfig: ProducerConfiguration, c
                 match pendingMessage.Callback with
                 | SingleCallback _ -> 1
                 | BatchCallbacks callbacks -> callbacks.Length
-            for _ in 1..messagesToRelease do
+            let mutable i = messagesToRelease
+            while i > 0 && blockedRequests.Count > 0 do
                 blockedRequests.Dequeue()
                 |> BeginSendMessage
                 |> post this.Mb
+                i <- i - 1
         pendingMessages.Dequeue() |> ignore
 
     let resendMessages (clientCnx: ClientCnx) =

diff --git a/src/Pulsar.Client/Pulsar.Client.fsproj b/src/Pulsar.Client/Pulsar.Client.fsproj
@@ -7,17 +7,17 @@
     <Title>Pulsar.Client</Title>
     <RootNamespace>Pulsar.Client</RootNamespace>
     <AssemblyName>Pulsar.Client</AssemblyName>
-    <Version>2.12.4</Version>
+    <Version>2.12.6</Version>
     <Company>F# community</Company>
     <Description>.NET client library for Apache Pulsar</Description>
     <RepositoryUrl>https://github.com/fsprojects/pulsar-client-dotnet</RepositoryUrl>
-    <PackageReleaseNotes>Packages update</PackageReleaseNotes>
+    <PackageReleaseNotes>Fix for unblocking correct amount of messages after receving acknowledge for a batch</PackageReleaseNotes>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <PackageProjectUrl>https://github.com/fsprojects/pulsar-client-dotnet</PackageProjectUrl>
     <RepositoryType>git</RepositoryType>
     <PackageTags>pulsar</PackageTags>
     <Authors>F# community</Authors>
-    <PackageVersion>2.12.4</PackageVersion>
+    <PackageVersion>2.12.6</PackageVersion>
     <DebugType>portable</DebugType>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <PackageReadmeFile>README.md</PackageReadmeFile>

diff --git a/tests/IntegrationTests/Batching.fs b/tests/IntegrationTests/Batching.fs
@@ -282,7 +282,7 @@ let tests =
 
             let client = getClient()
             let topicName = "public/default/topic-" + Guid.NewGuid().ToString("N")
-            let messagesNumber = 4
+            let messagesNumber = 5
 
             let! (consumer: IConsumer<byte[]>) =
                 client.NewConsumer()