Fix for BatchReceivePolicy.maxNumMessages not working #259

Author: Lanayx

src/Pulsar.Client/Common/DTO.fs

@@ -325,18 +325,16 @@ type Messages<'T> internal(maxNumberOfMessages: int, maxSizeOfMessages: int64) =
 
     let messageList = if maxNumberOfMessages > 0 then ResizeArray<Message<'T>>(maxNumberOfMessages) else ResizeArray<Message<'T>>()
 
-    member this.Count with get() =
-        currentNumberOfMessages
-    member this.Size with get() =
-        currentSizeOfMessages
+    member this.Count = currentNumberOfMessages
+    member this.Size = currentSizeOfMessages
 
-    member internal this.IsFull with get() =
+    member internal this.IsFull =
         currentNumberOfMessages = maxNumberOfMessages
         || currentSizeOfMessages = maxSizeOfMessages
 
     member internal this.CanAdd(message: Message<'T>) =
-        (maxNumberOfMessages > 0 && currentNumberOfMessages + 1 > maxNumberOfMessages)
-            || (maxSizeOfMessages > 0L && currentSizeOfMessages + (int64 message.Data.Length) > maxSizeOfMessages)
+        ((maxNumberOfMessages > 0 && currentNumberOfMessages + 1 > maxNumberOfMessages)
+            || (maxSizeOfMessages > 0L && currentSizeOfMessages + (int64 message.Data.Length) > maxSizeOfMessages))
         |> not
 
     member internal this.Add(message: Message<'T>) =

src/Pulsar.Client/Internal/ConsumerImpl.fs

@@ -528,7 +528,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                 messages.Add(interceptors.BeforeConsume(this, msg))
             else
                 shouldContinue <- false
-        Log.Logger.LogDebug("{0} BatchFormed with size {1}", prefix, messages.Size)
+        Log.Logger.LogDebug("{0} BatchFormed with size {1} and count {2}", prefix, messages.Size, messages.Count)
         channel.SetResult messages
 
     let removeExpiredMessagesFromQueue (msgIds: RedeliverSet) =

tests/UnitTests/Common/MessageTests.fs

@@ -1,5 +1,6 @@
 module Pulsar.Client.UnitTests.Common.MessageTests
 
+open System
 open Expecto
 open Expecto.Flip
 open Pulsar.Client.Common
@@ -64,4 +65,24 @@ let tests =
             let deserialized = MessageId.FromByteArray msgIdData
             Expect.equal "" msgId deserialized
         }
+
+        test "Message batching by count works correctly" {
+            let messages = Messages(2, -1)
+            let message = Message(MessageId.Earliest, [||], %"", false, EmptyProps, None, [||], %0L, [||], %0L, Nullable(), 0, "", "", fun () -> failwith "not implemented")
+            messages.CanAdd(message) |> Expect.isTrue ""
+            messages.Add(message)
+            messages.CanAdd(message) |> Expect.isTrue ""
+            messages.Add(message)
+            messages.CanAdd(message) |> Expect.isFalse ""
+        }
+
+        test "Message batching by size works correctly" {
+            let messages = Messages(-1, 2)
+            let message = Message(MessageId.Earliest, [| 0uy |], %"", false, EmptyProps, None, [||], %0L, [||], %0L, Nullable(), 0, "", "", fun () -> failwith "not implemented")
+            messages.CanAdd(message) |> Expect.isTrue ""
+            messages.Add(message)
+            messages.CanAdd(message) |> Expect.isTrue ""
+            messages.Add(message)
+            messages.CanAdd(message) |> Expect.isFalse ""
+        }
     ]