Added pause/resume for producers

Author: accelerated

include/cppkafka/consumer.h

@@ -202,7 +202,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
     /**
      * \brief Resumes all consumption
      */
-     void resume();
+    void resume();
 
     /**
      * \brief Commits the current partition assignment

include/cppkafka/producer.h

@@ -155,6 +155,20 @@ class CPPKAFKA_API Producer : public KafkaHandleBase {
      * \param timeout The timeout used on this call
      */
     void flush(std::chrono::milliseconds timeout);
+    
+    /**
+     * \brief Pauses production for this topic
+     *
+     * \param topic The topic name
+     */
+    void pause(const std::string& topic);
+    
+    /**
+     * \brief Resumes production for this topic
+     *
+     * \param topic The topic name
+     */
+    void resume(const std::string& topic);
 private:
     PayloadPolicy message_payload_policy_;
 };

include/cppkafka/topic_partition_list.h

@@ -37,6 +37,7 @@
 #include <set>
 #include <librdkafka/rdkafka.h>
 #include "macros.h"
+#include "metadata.h"
 
 namespace cppkafka {
 
@@ -53,6 +54,8 @@ using TopicPartitionList = std::vector<TopicPartition>;
 CPPKAFKA_API TopicPartitionsListPtr convert(const TopicPartitionList& topic_partitions);
 CPPKAFKA_API TopicPartitionList convert(const TopicPartitionsListPtr& topic_partitions);
 CPPKAFKA_API TopicPartitionList convert(rd_kafka_topic_partition_list_t* topic_partitions);
+CPPKAFKA_API TopicPartitionList convert(const std::string& topic,
+                                        const std::vector<PartitionMetadata>& partition_metadata);
 CPPKAFKA_API TopicPartitionsListPtr make_handle(rd_kafka_topic_partition_list_t* handle);
 
 // Extracts a partition list subset belonging to the provided topics (case-insensitive)

src/producer.cpp

@@ -114,4 +114,12 @@ void Producer::flush(milliseconds timeout) {
     check_error(result);
 }
 
+void Producer::pause(const std::string& topic) {
+    pause_partitions(convert(topic, get_metadata(get_topic(topic)).get_partitions()));
+}
+
+void Producer::resume(const std::string& topic) {
+    resume_partitions(convert(topic, get_metadata(get_topic(topic)).get_partitions()));
+}
+
 } // cppkafka

src/topic_partition_list.cpp

@@ -66,6 +66,16 @@ TopicPartitionList convert(rd_kafka_topic_partition_list_t* topic_partitions) {
     return output;
 }
 
+TopicPartitionList convert(const std::string& topic,
+                           const std::vector<PartitionMetadata>& partition_metadata)
+{
+    TopicPartitionList output;
+    for (const auto& meta : partition_metadata) {
+        output.emplace_back(topic, meta.get_id());
+    }
+    return output;
+}
+
 TopicPartitionsListPtr make_handle(rd_kafka_topic_partition_list_t* handle) {
     return TopicPartitionsListPtr(handle, &rd_kafka_topic_partition_list_destroy);
 }