Telecominfraproject/wlan-cloud-lib-cppkafka
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-cloud-lib-cppkafka.git synced 2026-01-27 10:22:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added pause/resume for producers (#87)

Browse Source 
* Added pause/resume for producers

* Moved pause/resume functions to KafkaHandleBase
This commit is contained in:
Alex Damian
2018-06-25 12:16:57 -04:00
committed by Matias Fontanini
parent 069ea3df8e
commit 5c72f3fe28
6 changed files with 39 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/kafka_handle_base.cpp
View File

@@ -64,6 +64,10 @@ void KafkaHandleBase::pause_partitions(const TopicPartitionList& topic_partition
check_error(error);
}
void KafkaHandleBase::pause(const std::string& topic) {
pause_partitions(convert(topic, get_metadata(get_topic(topic)).get_partitions()));
}
void KafkaHandleBase::resume_partitions(const TopicPartitionList& topic_partitions) {
TopicPartitionsListPtr topic_list_handle = convert(topic_partitions);
rd_kafka_resp_err_t error = rd_kafka_resume_partitions(get_handle(),
@@ -71,6 +75,10 @@ void KafkaHandleBase::resume_partitions(const TopicPartitionList& topic_partitio
check_error(error);
}
void KafkaHandleBase::resume(const std::string& topic) {
resume_partitions(convert(topic, get_metadata(get_topic(topic)).get_partitions()));
}
void KafkaHandleBase::set_timeout(milliseconds timeout) {
timeout_ms_ = timeout;
}

11
src/topic_partition_list.cpp
View File

@@ -32,6 +32,7 @@
#include "topic_partition_list.h"
#include "topic_partition.h"
#include "exceptions.h"
#include "metadata.h"
using std::vector;
using std::set;
@@ -66,6 +67,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);
}