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

Add commit methods to Consumer

Browse Source
This commit is contained in:
Matias Fontanini
2016-05-21 20:16:08 -07:00
parent 12621b11ad
commit 5606b81ecb
2 changed files with 19 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
include/cppkafka/consumer.h
View File

@@ -24,10 +24,14 @@ public:
void assign(const TopicPartitionList& topic_partitions); void assign(const TopicPartitionList& topic_partitions);
void commit(const Message& msg);
void async_commit(const Message& msg);
Message poll(); Message poll();
private: private:
static const std::chrono::milliseconds DEFAULT_TIMEOUT; static const std::chrono::milliseconds DEFAULT_TIMEOUT;
void commit(const Message& msg, bool async);
void check_error(rd_kafka_resp_err_t error); void check_error(rd_kafka_resp_err_t error);
std::chrono::milliseconds timeout_ms_; std::chrono::milliseconds timeout_ms_;

15
src/consumer.cpp
View File

@@ -45,11 +45,26 @@ void Consumer::assign(const TopicPartitionList& topic_partitions) {
check_error(error); check_error(error);
} }
void Consumer::commit(const Message& msg) {
commit(msg, false);
}
void Consumer::async_commit(const Message& msg) {
commit(msg, true);
}
Message Consumer::poll() { Message Consumer::poll() {
rd_kafka_message_t* message = rd_kafka_consumer_poll(get_handle(), timeout_ms_.count()); rd_kafka_message_t* message = rd_kafka_consumer_poll(get_handle(), timeout_ms_.count());
return Message(message); return Message(message);
} }
void Consumer::commit(const Message& msg, bool async) {
rd_kafka_resp_err_t error;
error = rd_kafka_commit_message(get_handle(), msg.get_handle(),
async ? 1 : 0);
check_error(error);
}
void Consumer::check_error(rd_kafka_resp_err_t error) { void Consumer::check_error(rd_kafka_resp_err_t error) {
if (error != RD_KAFKA_RESP_ERR_NO_ERROR) { if (error != RD_KAFKA_RESP_ERR_NO_ERROR) {
throw HandleException(error); throw HandleException(error);