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 legacy offset store API

Browse Source
This commit is contained in:
accelerated
2019-02-04 12:10:45 -05:00
parent 872ee0442b
commit 7bc03185a8
2 changed files with 54 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/consumer.cpp
View File

@@ -200,6 +200,28 @@ Consumer::get_offsets_position(const TopicPartitionList& topic_partitions) const
return convert(topic_list_handle);
}
void Consumer::store_offsets() const
{
rd_kafka_topic_partition_list_t* list = nullptr;
rd_kafka_resp_err_t error = rd_kafka_assignment(get_handle(), &list);
check_error(error);
error = rd_kafka_offsets_store(get_handle(), list);
check_error(error, list);
}
void Consumer::store_offsets(const TopicPartitionList& topic_partitions) const
{
TopicPartitionsListPtr topic_list_handle = convert(topic_partitions);
rd_kafka_resp_err_t error = rd_kafka_offsets_store(get_handle(), topic_list_handle.get());
check_error(error, topic_list_handle.get());
}
void Consumer::store_offset(const Message& msg) const
{
rd_kafka_resp_err_t error = rd_kafka_offset_store(msg.get_handle()->rkt, msg.get_partition(), msg.get_offset());
check_error(error);
}
vector<string> Consumer::get_subscription() const {
rd_kafka_resp_err_t error;
rd_kafka_topic_partition_list_t* list = nullptr;