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

Make Consumer::close private

Browse Source
This commit is contained in:
Matias Fontanini
2016-07-16 08:45:33 -07:00
parent 06c50f0f71
commit c3d6ce99f5
2 changed files with 6 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
include/cppkafka/consumer.h
View File

@@ -193,13 +193,6 @@ public:
*/ */
void unassign(); void unassign();
/**
* \brief Closes the consumer session
*
* This translates into a call to rd_kafka_consumer_close
*/
void close();
/** /**
* \brief Commits the given message synchronously * \brief Commits the given message synchronously
* *
@@ -323,6 +316,7 @@ private:
static void rebalance_proxy(rd_kafka_t *handle, rd_kafka_resp_err_t error, static void rebalance_proxy(rd_kafka_t *handle, rd_kafka_resp_err_t error,
rd_kafka_topic_partition_list_t *partitions, void *opaque); rd_kafka_topic_partition_list_t *partitions, void *opaque);
void close();
void commit(const Message& msg, bool async); void commit(const Message& msg, bool async);
void commit(const TopicPartitionList& topic_partitions, bool async); void commit(const TopicPartitionList& topic_partitions, bool async);
void handle_rebalance(rd_kafka_resp_err_t err, TopicPartitionList& topic_partitions); void handle_rebalance(rd_kafka_resp_err_t err, TopicPartitionList& topic_partitions);

10
src/consumer.cpp
View File

@@ -105,11 +105,6 @@ void Consumer::unassign() {
check_error(error); check_error(error);
} }
void Consumer::close() {
rd_kafka_resp_err_t error = rd_kafka_consumer_close(get_handle());
check_error(error);
}
void Consumer::commit(const Message& msg) { void Consumer::commit(const Message& msg) {
commit(msg, false); commit(msg, false);
} }
@@ -198,6 +193,11 @@ Message Consumer::poll() {
return message ? Message(message) : Message(); return message ? Message(message) : Message();
} }
void Consumer::close() {
rd_kafka_resp_err_t error = rd_kafka_consumer_close(get_handle());
check_error(error);
}
void Consumer::commit(const Message& msg, bool async) { void Consumer::commit(const Message& msg, bool async) {
rd_kafka_resp_err_t error; rd_kafka_resp_err_t error;
error = rd_kafka_commit_message(get_handle(), msg.get_handle(), error = rd_kafka_commit_message(get_handle(), msg.get_handle(),