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

Invoke error callback if present instead of log callback (#93)

Browse Source
This commit is contained in:
Alex Damian
2018-06-20 12:11:24 -04:00
committed by Matias Fontanini
parent eb46b8808e
commit c5aca985b8
2 changed files with 16 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
include/cppkafka/configuration.h
View File

@@ -62,19 +62,22 @@ class KafkaHandleBase;
class CPPKAFKA_API Configuration : public ConfigurationBase<Configuration> {
public:
using DeliveryReportCallback = std::function<void(Producer& producer, const Message&)>;
using OffsetCommitCallback = std::function<void(Consumer& consumer, Error,
using OffsetCommitCallback = std::function<void(Consumer& consumer,
Error error,
const TopicPartitionList& topic_partitions)>;
using ErrorCallback = std::function<void(KafkaHandleBase& handle, int error,
using ErrorCallback = std::function<void(KafkaHandleBase& handle,
int error,
const std::string& reason)>;
using ThrottleCallback = std::function<void(KafkaHandleBase& handle,
const std::string& broker_name,
int32_t broker_id,
std::chrono::milliseconds throttle_time)>;
using LogCallback = std::function<void(KafkaHandleBase& handle, int level,
using LogCallback = std::function<void(KafkaHandleBase& handle,
int level,
const std::string& facility,
const std::string& message)>;
using StatsCallback = std::function<void(KafkaHandleBase& handle, const std::string& json)>;
using SocketCallback = std::function<int(int domain, int type, int protoco)>;
using SocketCallback = std::function<int(int domain, int type, int protocol)>;
using ConfigurationBase<Configuration>::set;
using ConfigurationBase<Configuration>::get;

15
src/consumer.cpp
View File

@@ -79,16 +79,19 @@ Consumer::~Consumer() {
rebalance_error_callback_ = nullptr;
close();
}
catch (const Exception& ex) {
const char* library_name = "cppkafka";
catch (const HandleException& ex) {
ostringstream error_msg;
error_msg << "Failed to close consumer [" << get_name() << "]: " << ex.what();
CallbackInvoker<Configuration::LogCallback> logger("log", get_configuration().get_log_callback(), nullptr);
if (logger) {
logger(*this, static_cast<int>(LogLevel::LOG_ERR), library_name, error_msg.str());
CallbackInvoker<Configuration::ErrorCallback> error_cb("error", get_configuration().get_error_callback(), this);
CallbackInvoker<Configuration::LogCallback> logger_cb("log", get_configuration().get_log_callback(), nullptr);
if (error_cb) {
error_cb(*this, static_cast<int>(ex.get_error().get_error()), error_msg.str());
}
else if (logger_cb) {
logger_cb(*this, static_cast<int>(LogLevel::LOG_ERR), "cppkafka", error_msg.str());
}
else {
rd_kafka_log_print(get_handle(), static_cast<int>(LogLevel::LOG_ERR), library_name, error_msg.str().c_str());
rd_kafka_log_print(get_handle(), static_cast<int>(LogLevel::LOG_ERR), "cppkafka", error_msg.str().c_str());
}
}
}