scottk/wlan-cloud-lib-cppkafka
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-cloud-lib-cppkafka.git synced 2025-11-01 02:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add some other methods to TopicPartitionList

Browse Source
This commit is contained in:
Matias Fontanini
2016-05-21 20:06:54 -07:00
parent 1c51bb72f4
commit 12621b11ad
2 changed files with 35 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
include/cppkafka/topic_partition_list.h
View File

@@ -28,6 +28,10 @@ public:
TopicPartitionList& operator=(TopicPartitionList&&) = default;
void add(const TopicPartition& topic_partition);
void update(const TopicPartition& topic_partition);
bool remove(const TopicPartition& topic_partition);
bool contains(const TopicPartition& topic_partition) const;
size_t size() const;
bool empty() const;
@@ -40,6 +44,8 @@ private:
static HandlePtr make_handle(rd_kafka_topic_partition_list_t* ptr);
rd_kafka_topic_partition_t* get_topic_partition(const TopicPartition& topic_partition) const;
HandlePtr handle_;
};

29
src/topic_partition_list.cpp
View File

@@ -1,5 +1,6 @@
#include "topic_partition_list.h"
#include "topic_partition.h"
#include "exceptions.h"
namespace cppkafka {
@@ -33,6 +34,27 @@ void TopicPartitionList::add(const TopicPartition& topic_partition) {
element->offset = topic_partition.get_offset();
}
void TopicPartitionList::update(const TopicPartition& topic_partition) {
rd_kafka_resp_err_t error;
error = rd_kafka_topic_partition_list_set_offset(get_handle(),
topic_partition.get_topic().data(),
topic_partition.get_partition(),
topic_partition.get_offset());
if (error != RD_KAFKA_RESP_ERR_NO_ERROR) {
throw HandleException(error);
}
}
bool TopicPartitionList::remove(const TopicPartition& topic_partition) {
return rd_kafka_topic_partition_list_del(get_handle(),
topic_partition.get_topic().data(),
topic_partition.get_partition()) == 1;
}
bool TopicPartitionList::contains(const TopicPartition& topic_partition) const {
return get_topic_partition(topic_partition) != nullptr;
}
size_t TopicPartitionList::size() const {
return handle_->cnt;
}
@@ -50,4 +72,11 @@ TopicPartitionList::make_handle(rd_kafka_topic_partition_list_t* ptr) {
return HandlePtr(ptr, &rd_kafka_topic_partition_list_destroy);
}
rd_kafka_topic_partition_t*
TopicPartitionList::get_topic_partition(const TopicPartition& topic_partition) const {
return rd_kafka_topic_partition_list_find(get_handle(),
topic_partition.get_topic().data(),
topic_partition.get_partition());
}
} // cppkafka