Implement nullvalue

src/Pulsar.Client/Common/DTO.fs

@@ -213,6 +213,7 @@ type internal Metadata =
         EncryptionAlgo: string
         OrderingKey: byte[]
         ReplicatedFrom: string
+        NullValue: bool
     }
 
 type MessageKey =

src/Pulsar.Client/Internal/BatchMessageContainer.fs

@@ -46,6 +46,7 @@ module internal BatchHelpers =
                     if message.Properties.Count > 0 then
                         for property in message.Properties do
                             smm.Properties.Add(KeyValue(Key = property.Key, Value = property.Value))
+                    smm.NullValue <- box message.Value |> isNull
                     Serializer.SerializeWithLengthPrefix(messageStream, smm, PrefixStyle.Fixed32BigEndian)
                     messageWriter.Write(message.Payload)
                     struct(BatchDetails(%index, BatchMessageAcker.NullAcker), message, batchItem.Tcs)

src/Pulsar.Client/Internal/ClientCnx.fs

@@ -530,6 +530,7 @@ and internal ClientCnx (config: PulsarClientConfiguration,
             EncryptionAlgo = messageMetadata.EncryptionAlgo
             OrderingKey = messageMetadata.OrderingKey
             ReplicatedFrom = messageMetadata.ReplicatedFrom
+            NullValue = messageMetadata.NullValue
         }
 
         {

src/Pulsar.Client/Internal/ConsumerImpl.fs

@@ -668,9 +668,12 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
         else
             let msgKey = rawMessage.MessageKey
             let getValue () =
-                keyValueProcessor
-                |> Option.map (fun kvp -> kvp.DecodeKeyValue(msgKey, payload) :?> 'T)
-                |> Option.defaultWith (fun () -> schemaDecodeFunction payload)
+                if rawMessage.Metadata.NullValue then
+                    Unchecked.defaultof<'T>
+                else
+                    keyValueProcessor
+                        |> Option.map (fun kvp -> kvp.DecodeKeyValue(msgKey, payload) :?> 'T)
+                        |> Option.defaultWith (fun () -> schemaDecodeFunction payload)
             let message = Message(
                             msgId,
                             payload,
@@ -1330,9 +1333,12 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                 }
                 let msgKey = singleMessageMetadata.PartitionKey
                 let getValue () =
-                    keyValueProcessor
-                    |> Option.map (fun kvp -> kvp.DecodeKeyValue(msgKey, singleMessagePayload) :?> 'T)
-                    |> Option.defaultWith (fun() -> schemaDecodeFunction singleMessagePayload)
+                    if singleMessageMetadata.NullValue then
+                        Unchecked.defaultof<'T>
+                    else
+                        keyValueProcessor
+                            |> Option.map (fun kvp -> kvp.DecodeKeyValue(msgKey, singleMessagePayload) :?> 'T)
+                            |> Option.defaultWith (fun () -> schemaDecodeFunction singleMessagePayload)
                 let properties =
                     if singleMessageMetadata.Properties.Count > 0 then
                         singleMessageMetadata.Properties

src/Pulsar.Client/Internal/ProducerImpl.fs

@@ -857,7 +857,8 @@ type internal ProducerImpl<'T> private (producerConfig: ProducerConfiguration, c
                         Some { PartitionKey = %key; IsBase64Encoded = false }
                     else
                         Some { PartitionKey = %Convert.ToBase64String(keyBytes); IsBase64Encoded = true }
-                MessageBuilder(value, schema.Encode(value), keyObj,
+                let payloay = if box value |> isNull then Array.empty<byte> else schema.Encode(value)
+                MessageBuilder(value, payloay, keyObj,
                     ?properties0 = (properties |> Option.ofObj),
                     ?deliverAt = (deliverAt |> Option.ofNullable),
                     ?sequenceId = (sequenceId |> Option.ofNullable),

tests/IntegrationTests/Batching.fs

@@ -317,4 +317,45 @@ let tests =
             Log.Debug("Finished 'Second batch is formed well after the first one'")
 
         }
+
+        testTask "Null message with batch get sent if batch size exceeds" {
+
+            Log.Debug("Started 'Null message with batch get sent if batch size exceeds'")
+
+            let client = getClient()
+            let topicName = "public/default/topic-" + Guid.NewGuid().ToString("N")
+            let messagesNumber = 5
+
+            let! (consumer: IConsumer<byte[]>) =
+                client.NewConsumer()
+                    .Topic(topicName)
+                    .ConsumerName("batch consumer")
+                    .SubscriptionName("batch-subscription")
+                    .SubscribeAsync()
+
+            let! (producer: IProducer<byte[]>) =
+                client.NewProducer()
+                    .Topic(topicName)
+                    .ProducerName("batch producer")
+                    .EnableBatching(true)
+                    .BatchingMaxMessages(messagesNumber / 2)
+                    .BatchingMaxBytes(100)
+                    .MaxPendingMessages(1)
+                    .BlockIfQueueFull(true)
+                    .CreateAsync()
+
+            for i in 0 .. messagesNumber-1 do
+                producer.SendAsync(producer.NewMessage(null)) |> ignore
+
+            for i in 0 .. messagesNumber-1 do
+                let! (message: Message<byte[]>) = consumer.ReceiveAsync()
+                match message.MessageId.Type with
+                | Batch (index, _) ->
+                    Expect.equal $"Run {i} failed" (i % 2) %index
+                | _ ->
+                    failwith "Expected batch message"
+
+            Log.Debug("Finished 'Null message with batch get sent if batch size exceeds'")
+
+        }
     ]