Support Dead Letter Topic.

Author: shibd

include/pulsar/ConsumerConfiguration.h

@@ -34,6 +34,7 @@
 #include <memory>
 
 #include "BatchReceivePolicy.h"
+#include "DeadLetterPolicy.h"
 
 namespace pulsar {
 
@@ -398,6 +399,20 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     const BatchReceivePolicy& getBatchReceivePolicy() const;
 
+    /**
+     * Set dead letter policy.
+     *
+     * @param deadLetterPolicy thd default is empty
+     */
+    void setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy);
+
+    /**
+     * Get dead letter policy.
+     *
+     * @return dead letter policy
+     */
+    const DeadLetterPolicy& getDeadLetterPolicy() const;
+
     /**
      * Set whether the subscription status should be replicated.
      * The default value is `false`.

include/pulsar/DeadLetterPolicy.h

@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#ifndef DEAD_LETTER_POLICY_HPP_
+#define DEAD_LETTER_POLICY_HPP_
+
+#include <pulsar/defines.h>
+
+#include <memory>
+#include <string>
+
+namespace pulsar {
+
+struct DeadLetterPolicyImpl;
+
+/**
+ * Configuration for the "dead letter queue" feature in consumer.
+ *
+ * see @DeadLetterPolicyBuilder
+ */
+class PULSAR_PUBLIC DeadLetterPolicy {
+   public:
+    DeadLetterPolicy();
+
+    /**
+     * Get dead letter topic
+     *
+     * @return
+     */
+    std::string getDeadLetterTopic() const;
+
+    /**
+     * Get max redeliver count
+     *
+     * @return
+     */
+    int getMaxRedeliverCount() const;
+
+    /**
+     * Get initial subscription name
+     *
+     * @return
+     */
+    std::string getInitialSubscriptionName() const;
+
+   private:
+    friend class DeadLetterPolicyBuilder;
+
+    typedef std::shared_ptr<DeadLetterPolicyImpl> DeadLetterPolicyImplPtr;
+    DeadLetterPolicyImplPtr impl_;
+
+    explicit DeadLetterPolicy(const DeadLetterPolicyImplPtr& impl);
+};
+}  // namespace pulsar
+
+#endif /* DEAD_LETTER_POLICY_HPP_ */

include/pulsar/DeadLetterPolicyBuilder.h

@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#ifndef DEAD_LETTER_POLICY_BUILD_HPP_
+#define DEAD_LETTER_POLICY_BUILD_HPP_
+
+#include <pulsar/DeadLetterPolicy.h>
+#include <pulsar/defines.h>
+
+#include <memory>
+
+namespace pulsar {
+
+struct DeadLetterPolicyImpl;
+
+/**
+ * The builder to build a DeadLetterPolicyBuilder
+ *
+ * Example of building DeadLetterPolicy:
+ *
+ * ```c++
+ * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder()
+ *                       .deadLetterTopic("dlq-topic")
+ *                       .maxRedeliverCount(10)
+ *                       .initialSubscriptionName("init-sub-name")
+ *                       .build();
+ * ```
+ */
+class PULSAR_PUBLIC DeadLetterPolicyBuilder {
+   public:
+    DeadLetterPolicyBuilder();
+
+    /**
+     * Set dead letter topic
+     *
+     * @return
+     */
+    DeadLetterPolicyBuilder& deadLetterTopic(const std::string& deadLetterTopic);
+
+    /**
+     * Set max redeliver count
+     *
+     * @return
+     */
+    DeadLetterPolicyBuilder& maxRedeliverCount(int maxRedeliverCount);
+
+    /**
+     * Set initial subscription name
+     *
+     * @return
+     */
+    DeadLetterPolicyBuilder& initialSubscriptionName(const std::string& initialSubscriptionName);
+
+    /**
+     * Build DeadLetterPolicy.
+     *
+     * @return
+     */
+    DeadLetterPolicy build();
+
+   private:
+    std::shared_ptr<DeadLetterPolicyImpl> impl_;
+};
+}  // namespace pulsar
+
+#endif /* DEAD_LETTER_POLICY_BUILD_HPP_ */

include/pulsar/ProducerConfiguration.h

@@ -532,6 +532,18 @@ class PULSAR_PUBLIC ProducerConfiguration {
      */
     ProducerAccessMode getAccessMode() const;
 
+    /**
+     * Use this configuration to automatically create an initial subscription when creating a topic.
+     *
+     * If this field is not set, the initial subscription is not created.
+     */
+    ProducerConfiguration& setInitialSubscriptionName(const std::string& initialSubscriptionName);
+
+    /**
+     * Get initial subscription name.
+     */
+    const std::string& getInitialSubscriptionName() const;
+
     friend class PulsarWrapper;
 
    private:

lib/Commands.cc

@@ -385,7 +385,8 @@ SharedBuffer Commands::newProducer(const std::string& topic, uint64_t producerId
                                    const std::map<std::string, std::string>& metadata,
                                    const SchemaInfo& schemaInfo, uint64_t epoch,
                                    bool userProvidedProducerName, bool encrypted,
-                                   ProducerAccessMode accessMode, boost::optional<uint64_t> topicEpoch) {
+                                   ProducerAccessMode accessMode, boost::optional<uint64_t> topicEpoch,
+                                   std::string initialSubscriptionName) {
     BaseCommand cmd;
     cmd.set_type(BaseCommand::PRODUCER);
     CommandProducer* producer = cmd.mutable_producer();
@@ -399,6 +400,9 @@ SharedBuffer Commands::newProducer(const std::string& topic, uint64_t producerId
     if (topicEpoch) {
         producer->set_topic_epoch(topicEpoch.value());
     }
+    if (!initialSubscriptionName.empty()) {
+        producer->set_initial_subscription_name(initialSubscriptionName);
+    }
 
     for (std::map<std::string, std::string>::const_iterator it = metadata.begin(); it != metadata.end();
          it++) {

lib/Commands.h

@@ -110,7 +110,8 @@ class Commands {
                                     const std::map<std::string, std::string>& metadata,
                                     const SchemaInfo& schemaInfo, uint64_t epoch,
                                     bool userProvidedProducerName, bool encrypted,
-                                    ProducerAccessMode accessMode, boost::optional<uint64_t> topicEpoch);
+                                    ProducerAccessMode accessMode, boost::optional<uint64_t> topicEpoch,
+                                    std::string initialSubscriptionName);
 
     static SharedBuffer newAck(uint64_t consumerId, int64_t ledgerId, int64_t entryId,
                                CommandAck_AckType ackType, CommandAck_ValidationError validationError);

lib/ConsumerConfiguration.cc

@@ -287,4 +287,10 @@ const BatchReceivePolicy& ConsumerConfiguration::getBatchReceivePolicy() const {
     return impl_->batchReceivePolicy;
 }
 
+void ConsumerConfiguration::setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy) {
+    impl_->deadLetterPolicy = deadLetterPolicy;
+}
+
+const DeadLetterPolicy& ConsumerConfiguration::getDeadLetterPolicy() const { return impl_->deadLetterPolicy; }
+
 }  // namespace pulsar

lib/ConsumerConfigurationImpl.h

@@ -46,6 +46,7 @@ struct ConsumerConfigurationImpl {
     bool readCompacted{false};
     InitialPosition subscriptionInitialPosition{InitialPosition::InitialPositionLatest};
     BatchReceivePolicy batchReceivePolicy{};
+    DeadLetterPolicy deadLetterPolicy;
     int patternAutoDiscoveryPeriod{60};
     bool replicateSubscriptionStateEnabled{false};
     std::map<std::string, std::string> properties;