Add dlq initial subscription support

src/Pulsar.Client/Api/Configuration.fs

@@ -138,6 +138,7 @@ type ProducerConfiguration =
         BlockIfQueueFull: bool
         MessageEncryptor: IMessageEncryptor option
         ProducerCryptoFailureAction: ProducerCryptoFailureAction
+        InitialSubscriptionName: string
     }
     member this.BatchingPartitionSwitchFrequencyIntervalMs =
         this.BatchingPartitionSwitchFrequencyByPublishDelay * (int this.BatchingMaxPublishDelay.TotalMilliseconds)
@@ -165,6 +166,7 @@ type ProducerConfiguration =
             BlockIfQueueFull = false
             MessageEncryptor = None
             ProducerCryptoFailureAction = ProducerCryptoFailureAction.FAIL
+            InitialSubscriptionName = ""
         }
 
 type ReaderConfiguration =

src/Pulsar.Client/Api/ConsumerBuilder.fs

@@ -18,9 +18,10 @@ type ConsumerBuilder<'T> private (createConsumerAsync, createProducerAsync, conf
     let deadLettersProcessor (c: ConsumerConfiguration<'T>) (deadLettersPolicy: DeadLetterPolicy) (topic: TopicName)  =
         let getTopicName () =
             topic.ToString()
-        let createProducer deadLetterTopic =
+        let createProducer deadLetterTopic initialSubscriptionName =
             ProducerBuilder(createProducerAsync, schema)
                 .Topic(deadLetterTopic)
+                .InitialSubscriptionName(initialSubscriptionName)
                 .BlockIfQueueFull(false)
                 .CreateAsync()
         DeadLetterProcessor(deadLettersPolicy, getTopicName, c.SubscriptionName, createProducer) :> IDeadLetterProcessor<'T>
@@ -62,19 +63,21 @@ type ConsumerBuilder<'T> private (createConsumerAsync, createProducerAsync, conf
                     let prefixPart = c.SingleTopic.ToString() + "-" + %c.SubscriptionName
                     let defaultRetryLetterTopic = prefixPart + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX
                     let defaultDeadLetterTopic = prefixPart + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX
+                    let defaultInitialSubscriptionName = ""
                     let newPolicy =
                         match c.DeadLetterPolicy with
                         | None ->
-                            DeadLetterPolicy(RetryMessageUtil.MAX_RECONSUMETIMES, defaultDeadLetterTopic, defaultRetryLetterTopic)
+                            DeadLetterPolicy(RetryMessageUtil.MAX_RECONSUMETIMES, defaultDeadLetterTopic, defaultRetryLetterTopic, defaultInitialSubscriptionName)
                         | Some policy ->
+                            let initialSubscriptionName = if String.IsNullOrEmpty policy.InitialSubscriptionName then defaultInitialSubscriptionName else policy.InitialSubscriptionName 
                             let isEmptyDL = String.IsNullOrEmpty policy.DeadLetterTopic
                             let isEmptyRL = String.IsNullOrEmpty policy.RetryLetterTopic
                             if isEmptyDL && isEmptyRL then
-                                DeadLetterPolicy(policy.MaxRedeliveryCount, defaultDeadLetterTopic, defaultRetryLetterTopic)
+                                DeadLetterPolicy(policy.MaxRedeliveryCount, defaultDeadLetterTopic, defaultRetryLetterTopic, initialSubscriptionName)
                             elif isEmptyDL then
-                                DeadLetterPolicy(policy.MaxRedeliveryCount, defaultDeadLetterTopic, policy.RetryLetterTopic)
+                                DeadLetterPolicy(policy.MaxRedeliveryCount, defaultDeadLetterTopic, policy.RetryLetterTopic, initialSubscriptionName)
                             elif isEmptyRL then
-                                DeadLetterPolicy(policy.MaxRedeliveryCount, policy.DeadLetterTopic, defaultRetryLetterTopic)
+                                DeadLetterPolicy(policy.MaxRedeliveryCount, policy.DeadLetterTopic, defaultRetryLetterTopic, initialSubscriptionName)
                             else
                                 policy
                     { c with

src/Pulsar.Client/Api/DeadLetters.fs

@@ -8,10 +8,12 @@ open System.Threading.Tasks
 type DeadLetterPolicy(maxRedeliveryCount: int
                        , [<Optional; DefaultParameterValue(null:string)>] deadLetterTopic: string
                        , [<Optional; DefaultParameterValue(null:string)>] retryLetterTopic: string
+                       , [<Optional; DefaultParameterValue(null:string)>] initialSubscriptionName: string
                        ) =
     member __.MaxRedeliveryCount = maxRedeliveryCount
     member __.DeadLetterTopic = deadLetterTopic
     member __.RetryLetterTopic = retryLetterTopic
+    member __.InitialSubscriptionName = initialSubscriptionName
 
 type IDeadLetterProcessor<'T> =
     abstract member ClearMessages: unit -> unit

src/Pulsar.Client/Api/ProducerBuilder.fs

@@ -156,6 +156,11 @@ type ProducerBuilder<'T> private (сreateProducerAsync, config: ProducerConfigur
             MessageEncryptor = Some messageEncryptor }
         |> this.With
 
+    member internal this.InitialSubscriptionName initialSubscriptionName =
+        { config with
+            InitialSubscriptionName = initialSubscriptionName }
+        |> this.With
+
     member this.CreateAsync(): Task<IProducer<'T>> =
         сreateProducerAsync(verify config, schema, producerInterceptors)
 

src/Pulsar.Client/Common/Commands.fs

@@ -200,12 +200,14 @@ let newLookup (topicName : CompleteTopicName) (requestId : RequestId) (authorita
     command |> serializeSimpleCommand
 
 let newProducer (topicName : CompleteTopicName) (producerName: string) (producerId : ProducerId) (requestId : RequestId)
-                (schemaInfo: SchemaInfo) (epoch: uint64) (txnEnabled: bool) =
+                (schemaInfo: SchemaInfo) (epoch: uint64) (txnEnabled: bool) (initialSubscriptionName: string) =
     let schema = getProtoSchema schemaInfo
     let request = CommandProducer(Topic = %topicName, ProducerId = %producerId, RequestId = %requestId,
                                   Epoch = epoch, TxnEnabled = txnEnabled)
     if producerName |> String.IsNullOrEmpty |> not then
         request.ProducerName <- producerName
+    if initialSubscriptionName |> String.IsNullOrEmpty |> not then
+        request.InitialSubscriptionName <- initialSubscriptionName
     if schema.``type`` <> Schema.Type.None then
         request.Schema <- schema
     let command = BaseCommand(``type`` = CommandType.Producer, Producer = request)

src/Pulsar.Client/Internal/DeadLetters.fs

@@ -13,7 +13,7 @@ type internal DeadLetterProcessor<'T>
     (policy: DeadLetterPolicy,
      getTopicName: unit -> string,
      subscriptionName: SubscriptionName,
-     createProducer: string -> Task<IProducer<'T>>) =
+     createProducer: string -> string -> Task<IProducer<'T>>) =
 
     let topicName = getTopicName()
     let store = Dictionary<MessageId, Message<'T>>()
@@ -24,11 +24,11 @@ type internal DeadLetterProcessor<'T>
             $"{topicName}-{subscriptionName}{RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX}"
 
     let dlProducer = lazy (
-        createProducer dlTopicName
+        createProducer dlTopicName policy.InitialSubscriptionName
     )
 
     let rlProducer = lazy (
-        createProducer policy.RetryLetterTopic
+        createProducer policy.RetryLetterTopic ""
     )
 
     let getOptionalKey (message: Message<'T>) =

src/Pulsar.Client/Internal/ProducerImpl.fs

@@ -512,7 +512,7 @@ type internal ProducerImpl<'T> private (producerConfig: ProducerConfiguration, c
                     let requestId = Generators.getNextRequestId()
                     try
                         let payload = Commands.newProducer producerConfig.Topic.CompleteTopicName producerConfig.ProducerName
-                                            producerId requestId schema.SchemaInfo epoch clientConfig.EnableTransaction
+                                            producerId requestId schema.SchemaInfo epoch clientConfig.EnableTransaction producerConfig.InitialSubscriptionName
                         let! response = clientCnx.SendAndWaitForReply requestId payload
                         let success = response |> PulsarResponseType.GetProducerSuccess
                         if String.IsNullOrEmpty producerName then