Changes as per code review

Author: accelerated

Diff for: examples/CMakeLists.txt

@@ -1,4 +1,4 @@
-link_libraries(cppkafka ${RDKAFKA_LIBRARY} ${Boost_LIBRARIES} pthread)
+link_libraries(cppkafka ${RDKAFKA_LIBRARY} ${Boost_LIBRARIES} pthread rt ssl crypto dl z)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/../include)
 include_directories(SYSTEM ${RDKAFKA_INCLUDE_DIR})
 

Diff for: include/cppkafka/callback_invoker.h

@@ -37,19 +37,9 @@
 
 namespace cppkafka {
 
-// Success values
-template <typename T>
-T success_value() { return T(); }
-
-template<> inline
-void success_value<void>() {};
-
-template<> inline
-bool success_value<bool>() { return true; }
-
 // Error values
 template <typename T>
-T error_value() { return T(); }
+T error_value() { return T{}; }
 
 template<> inline
 void error_value<void>() {};
@@ -60,34 +50,40 @@ bool error_value<bool>() { return false; }
 template<> inline
 int error_value<int>() { return -1; }
 
-// CallbackInvoker
-template <typename RetType, typename ...Args>
-class CallbackInvoker;
-
-template <typename RetType, typename ...Args>
-class CallbackInvoker<RetType(Args...)>
+/**
+ * \brief Wraps an std::function object and runs it while preventing all exceptions from escaping
+ * \tparam Func An std::function object
+ */
+template <typename Func>
+class CallbackInvoker
 {
 public:
-    using Func = std::function<RetType(Args...)>;
+    using RetType = typename Func::result_type;
     using LogCallback = std::function<void(KafkaHandleBase& handle,
                                            int level,
                                            const std::string& facility,
                                            const std::string& message)>;
-    CallbackInvoker(const std::string& callback_name,
+    CallbackInvoker(const char* callback_name,
                     const Func& callback,
                     KafkaHandleBase* handle)
     : callback_name_(callback_name),
       callback_(callback),
       handle_(handle) {
     }
-    RetType operator()(Args... args) const {
+    
+    explicit operator bool() const {
+        return (bool)callback_;
+    }
+    
+    template <typename ...Args>
+    RetType operator()(Args&&... args) const {
         static const char* library_name = "cppkafka";
         std::ostringstream error_msg;
         try {
             if (callback_) {
                 return callback_(std::forward<Args>(args)...);
             }
-            return success_value<RetType>();
+            return error_value<RetType>();
         }
         catch (const std::exception& ex) {
             if (handle_) {
@@ -112,8 +108,10 @@ class CallbackInvoker<RetType(Args...)>
                 catch (...) {} // sink everything
             }
             else {
-                rd_kafka_log_print(handle_->get_handle(), static_cast<int>(LogLevel::LOG_ERR),
-                                   library_name, error_msg.str().c_str());
+                rd_kafka_log_print(handle_->get_handle(),
+                                   static_cast<int>(LogLevel::LOG_ERR),
+                                   library_name,
+                                   error_msg.str().c_str());
             }
         }
         return error_value<RetType>();

Diff for: include/cppkafka/utils/backoff_committer.h

@@ -128,15 +128,15 @@ class BackoffCommitter : public BackoffPerformer {
             return true;
         }
         catch (const HandleException& ex) {
-            static const std::string callback_name("backoff committer");
             // If there were actually no offsets to commit, return. Retrying won't solve
             // anything here
             if (ex.get_error() == RD_KAFKA_RESP_ERR__NO_OFFSET) {
                 return true;
             }
             // If there's a callback and it returns false for this message, abort.
             // Otherwise keep committing.
-            return !CallbackInvoker<bool(Error)>(callback_name, callback_, &consumer_)(ex.get_error());
+            CallbackInvoker<ErrorCallback> callback("backoff committer", callback_, &consumer_);
+            return callback && !callback(ex.get_error());
         }
     }
 

Diff for: include/cppkafka/utils/buffered_producer.h

@@ -364,9 +364,9 @@ void BufferedProducer<BufferType>::flush() {
             produce_message(flush_queue.front());
         }
         catch (const HandleException& ex) {
-            if (flush_failure_callback_ &&
-                flush_failure_callback_(flush_queue.front(), ex.get_error())) {
-                // retry again later
+            // If we have a flush failure callback and it returns true, we retry producing this message later
+            CallbackInvoker<FlushFailureCallback> callback("flush failure", flush_failure_callback_, &producer_);
+            if (callback && callback(flush_queue.front(), ex.get_error())) {
                 do_add_message(std::move(flush_queue.front()), MessagePriority::Low, false);
             }
         }
@@ -515,23 +515,22 @@ Configuration BufferedProducer<BufferType>::prepare_configuration(Configuration
 
 template <typename BufferType>
 void BufferedProducer<BufferType>::on_delivery_report(const Message& message) {
-    static const std::string ds_callback_name("delivery success");
-    static const std::string pf_callback_name("produce failure");
     // Decrement the expected acks
     --pending_acks_;
     assert(pending_acks_ != (size_t)-1); // Prevent underflow
-    // We should produce this message again if it has an error and we either don't have a
-    // produce failure callback or we have one but it returns true
-    bool should_produce = message.get_error() &&
-                          CallbackInvoker<bool(const Message&)>
-                              (pf_callback_name, produce_failure_callback_, &producer_)(message);
-    if (should_produce) {
-        // Re-enqueue for later retransmission with higher priority (i.e. front of the queue)
-        do_add_message(Builder(message), MessagePriority::High, false);
+    
+    if (message.get_error()) {
+        // We should produce this message again if we don't have a produce failure callback
+        // or we have one but it returns true
+        CallbackInvoker<ProduceFailureCallback> callback("produce failure", produce_failure_callback_, &producer_);
+        if (!callback || callback(message)) {
+            // Re-enqueue for later retransmission with higher priority (i.e. front of the queue)
+            do_add_message(Builder(message), MessagePriority::High, false);
+        }
     }
     else {
         // Successful delivery
-        CallbackInvoker<void(const Message&)>(ds_callback_name, produce_success_callback_, &producer_)(message);
+        CallbackInvoker<ProduceSuccessCallback>("delivery success", produce_success_callback_, &producer_)(message);
         // Increment the total successful transmissions
         ++total_messages_produced_;
     }

Diff for: src/configuration.cpp

@@ -50,68 +50,58 @@ namespace cppkafka {
 // Callback proxies
 
 void delivery_report_callback_proxy(rd_kafka_t*, const rd_kafka_message_t* msg, void *opaque) {
-    static const string callback_name("delivery report");
     Producer* handle = static_cast<Producer*>(opaque);
     Message message = Message::make_non_owning((rd_kafka_message_t*)msg);
-    CallbackInvoker<void(Producer&, const Message&)>
-        (callback_name, handle->get_configuration().get_delivery_report_callback(), handle)
+    CallbackInvoker<Configuration::DeliveryReportCallback>
+        ("delivery report", handle->get_configuration().get_delivery_report_callback(), handle)
         (*handle, message);
 }
 
 void offset_commit_callback_proxy(rd_kafka_t*, rd_kafka_resp_err_t err,
                                   rd_kafka_topic_partition_list_t *offsets, void *opaque) {
-    static const string callback_name("offset commit");
     Consumer* handle = static_cast<Consumer*>(opaque);
     TopicPartitionList list = offsets ? convert(offsets) : TopicPartitionList{};
-    CallbackInvoker<void(Consumer&, Error, const TopicPartitionList&)>
-        (callback_name, handle->get_configuration().get_offset_commit_callback(), handle)
+    CallbackInvoker<Configuration::OffsetCommitCallback>
+        ("offset commit", handle->get_configuration().get_offset_commit_callback(), handle)
         (*handle, err, list);
 }
 
 void error_callback_proxy(rd_kafka_t*, int err, const char *reason, void *opaque) {
-    static const string callback_name("error");
     KafkaHandleBase* handle = static_cast<KafkaHandleBase*>(opaque);
-    CallbackInvoker<void(KafkaHandleBase&, int, const std::string&)>
-        (callback_name, handle->get_configuration().get_error_callback(), handle)
+    CallbackInvoker<Configuration::ErrorCallback>
+        ("error", handle->get_configuration().get_error_callback(), handle)
         (*handle, err, reason);
 }
 
 void throttle_callback_proxy(rd_kafka_t*, const char* broker_name,
                               int32_t broker_id, int throttle_time_ms, void *opaque) {
-    static const string callback_name("throttle");
     KafkaHandleBase* handle = static_cast<KafkaHandleBase*>(opaque);
-     CallbackInvoker<void(KafkaHandleBase&, const std::string&, int32_t, std::chrono::milliseconds)>
-         (callback_name, handle->get_configuration().get_throttle_callback(), handle)
+     CallbackInvoker<Configuration::ThrottleCallback>
+         ("throttle", handle->get_configuration().get_throttle_callback(), handle)
          (*handle, broker_name, broker_id, milliseconds(throttle_time_ms));
 }
 
 void log_callback_proxy(const rd_kafka_t* h, int level,
                         const char* facility, const char* message) {
     KafkaHandleBase* handle = static_cast<KafkaHandleBase*>(rd_kafka_opaque(h));
-    const auto& callback = handle->get_configuration().get_log_callback();
-    if (callback) {
-        callback(*handle, level, facility, message);
-    }
+    CallbackInvoker<Configuration::LogCallback>
+        ("log", handle->get_configuration().get_log_callback(), nullptr)
+        (*handle, level, facility, message);
 }
 
 int stats_callback_proxy(rd_kafka_t*, char *json, size_t json_len, void *opaque) {
-    static const string callback_name("statistics");
     KafkaHandleBase* handle = static_cast<KafkaHandleBase*>(opaque);
-    CallbackInvoker<void(KafkaHandleBase&, const std::string&)>
-        (callback_name, handle->get_configuration().get_stats_callback(), handle)
+    CallbackInvoker<Configuration::StatsCallback>
+        ("statistics", handle->get_configuration().get_stats_callback(), handle)
         (*handle, string(json, json + json_len));
     return 0;
 }
 
 int socket_callback_proxy(int domain, int type, int protocol, void* opaque) {
-    static const string callback_name("socket");
     KafkaHandleBase* handle = static_cast<KafkaHandleBase*>(opaque);
-    if (handle->get_configuration().get_socket_callback()) {
-        return CallbackInvoker<int(int, int, int)>
-            (callback_name, handle->get_configuration().get_socket_callback(), handle)
-            (domain, type, protocol);
-    }
-    return -1;
+    return CallbackInvoker<Configuration::SocketCallback>
+        ("socket", handle->get_configuration().get_socket_callback(), handle)
+        (domain, type, protocol);
 }
 
 // Configuration

Diff for: src/consumer.cpp

@@ -80,12 +80,12 @@ Consumer::~Consumer() {
         close();
     }
     catch (const Exception& ex) {
-        constexpr const char* library_name = "cppkafka";
+        const char* library_name = "cppkafka";
         ostringstream error_msg;
         error_msg << "Failed to close consumer [" << get_name() << "]: " << ex.what();
-        const auto& callback = get_configuration().get_log_callback();
-        if (callback) {
-            callback(*this, static_cast<int>(LogLevel::LOG_ERR), library_name, error_msg.str());
+        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());
         }
         else {
             rd_kafka_log_print(get_handle(), static_cast<int>(LogLevel::LOG_ERR), library_name, error_msg.str().c_str());
@@ -283,18 +283,15 @@ void Consumer::commit(const TopicPartitionList* topic_partitions, bool async) {
 void Consumer::handle_rebalance(rd_kafka_resp_err_t error,
                                 TopicPartitionList& topic_partitions) {
     if (error == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) {
-        static const string callback_name("assignment");
-        CallbackInvoker<void(TopicPartitionList&)>(callback_name, assignment_callback_, this)(topic_partitions);
+        CallbackInvoker<AssignmentCallback>("assignment", assignment_callback_, this)(topic_partitions);
         assign(topic_partitions);
     }
     else if (error == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) {
-        static const string callback_name("revocation");
-        CallbackInvoker<void(const TopicPartitionList&)>(callback_name, revocation_callback_, this)(topic_partitions);
+        CallbackInvoker<RevocationCallback>("revocation", revocation_callback_, this)(topic_partitions);
         unassign();
     }
     else {
-        static const string callback_name("rebalance error");
-        CallbackInvoker<void(Error)>(callback_name, rebalance_error_callback_, this)(error);
+        CallbackInvoker<RebalanceErrorCallback>("rebalance error", rebalance_error_callback_, this)(error);
         unassign();
     }
 }

Diff for: src/topic_configuration.cpp

@@ -48,11 +48,9 @@ int32_t partitioner_callback_proxy(const rd_kafka_topic_t* handle, const void *k
     const TopicConfiguration* config = static_cast<TopicConfiguration*>(topic_opaque);
     const auto& callback = config->get_partitioner_callback();
     if (callback) {
-        static const string callback_name("topic partitioner");
         Topic topic = Topic::make_non_owning(const_cast<rd_kafka_topic_t*>(handle));
         Buffer key(static_cast<const char*>(key_ptr), key_size);
-        return CallbackInvoker<int32_t(const Topic&, const Buffer&, int32_t)>
-            (callback_name, callback, nullptr)
+        return CallbackInvoker<TopicConfiguration::PartitionerCallback>("topic partitioner", callback, nullptr)
             (topic, key, partition_count);
     }
     else {

Diff for: tests/CMakeLists.txt

@@ -8,7 +8,7 @@ add_custom_target(tests)
 
 include_directories(${CMAKE_CURRENT_SOURCE_DIR})
 add_library(cppkafka-test EXCLUDE_FROM_ALL test_utils.cpp)
-target_link_libraries(cppkafka-test cppkafka ${RDKAFKA_LIBRARY} pthread)
+target_link_libraries(cppkafka-test cppkafka ${RDKAFKA_LIBRARY} pthread rt ssl crypto dl z)
 
 add_definitions("-DKAFKA_TEST_INSTANCE=\"${KAFKA_TEST_INSTANCE}\"")