Added logic to conditionally enable internal data

Author: accelerated

Diff for: include/cppkafka/cppkafka.h

@@ -44,6 +44,7 @@
 #include <cppkafka/macros.h>
 #include <cppkafka/message.h>
 #include <cppkafka/message_builder.h>
+#include <cppkafka/message_internal.h>
 #include <cppkafka/metadata.h>
 #include <cppkafka/producer.h>
 #include <cppkafka/queue.h>

Diff for: include/cppkafka/message_internal.h

@@ -35,6 +35,8 @@
 
 namespace cppkafka {
 
+class Producer;
+
 struct Internal {
     virtual ~Internal() = default;
 };
@@ -44,25 +46,11 @@ using InternalPtr = std::shared_ptr<Internal>;
  * \brief Private message data structure
  */
 class MessageInternal {
-    friend class Producer;
-    
+    friend Producer;
 public:
-    static std::unique_ptr<MessageInternal> load(Message& message) {
-        if (message.get_user_data()) {
-            // Unpack internal data
-            std::unique_ptr<MessageInternal> internal_data(static_cast<MessageInternal*>(message.get_user_data()));
-            message.load_internal(internal_data->user_data_, internal_data->internal_);
-            return internal_data;
-        }
-        return nullptr;
-    }
-    
+    static std::unique_ptr<MessageInternal> load(const Producer& producer, Message& message);
 private:
-    MessageInternal(void* user_data, std::shared_ptr<Internal> internal)
-    : user_data_(user_data),
-      internal_(internal) {
-    }
-    
+    MessageInternal(void* user_data, std::shared_ptr<Internal> internal);
     void*          user_data_;
     InternalPtr    internal_;
 };

Diff for: include/cppkafka/producer.h

@@ -31,12 +31,14 @@
 #define CPPKAFKA_PRODUCER_H
 
 #include <memory>
+#include <tuple>
 #include "kafka_handle_base.h"
 #include "configuration.h"
 #include "buffer.h"
 #include "topic.h"
 #include "macros.h"
 #include "message_builder.h"
+#include "message_internal.h"
 
 namespace cppkafka {
 
@@ -78,6 +80,7 @@ class Message;
  */
 class CPPKAFKA_API Producer : public KafkaHandleBase {
 public:
+    friend MessageInternal;
     /**
      * The policy to use for the payload. The default policy is COPY_PAYLOAD
      */
@@ -156,7 +159,11 @@ class CPPKAFKA_API Producer : public KafkaHandleBase {
      */
     void flush(std::chrono::milliseconds timeout);
 private:
+    using LoadResult = std::tuple<void*, std::unique_ptr<MessageInternal>>;
+    LoadResult load_internal(void* user_data, InternalPtr internal);
+    
     PayloadPolicy message_payload_policy_;
+    bool has_internal_data_;
 };
 
 } // cppkafka

Diff for: src/CMakeLists.txt

@@ -7,6 +7,7 @@ set(SOURCES
     buffer.cpp
     queue.cpp
     message.cpp
+    message_internal.cpp
     topic_partition.cpp
     topic_partition_list.cpp
     metadata.cpp

Diff for: src/configuration.cpp

@@ -52,7 +52,7 @@ namespace cppkafka {
 void delivery_report_callback_proxy(rd_kafka_t*, const rd_kafka_message_t* msg, void *opaque) {
     Producer* handle = static_cast<Producer*>(opaque);
     Message message = Message::make_non_owning((rd_kafka_message_t*)msg);
-    unique_ptr<MessageInternal> internal_data(MessageInternal::load(message));
+    unique_ptr<MessageInternal> internal_data(MessageInternal::load(*handle, message));
     CallbackInvoker<Configuration::DeliveryReportCallback>
         ("delivery report", handle->get_configuration().get_delivery_report_callback(), handle)
         (*handle, message);

Diff for: src/message_internal.cpp

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2017, Matias Fontanini
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following disclaimer
+ *   in the documentation and/or other materials provided with the
+ *   distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+#include "message_internal.h"
+#include "producer.h"
+
+namespace cppkafka {
+
+MessageInternal::MessageInternal(void* user_data, std::shared_ptr<Internal> internal)
+: user_data_(user_data),
+  internal_(internal) {
+}
+
+std::unique_ptr<MessageInternal> MessageInternal::load(const Producer& producer, Message& message) {
+    if (producer.has_internal_data_ && message.get_user_data()) {
+        // Unpack internal data
+        std::unique_ptr<MessageInternal> internal_data(static_cast<MessageInternal*>(message.get_user_data()));
+        message.load_internal(internal_data->user_data_, internal_data->internal_);
+        return internal_data;
+    }
+    return nullptr;
+}
+
+}

Diff for: src/producer.cpp

@@ -37,11 +37,13 @@ using std::move;
 using std::string;
 using std::chrono::milliseconds;
 using std::unique_ptr;
+using std::get;
 
 namespace cppkafka {
 
 Producer::Producer(Configuration config)
-: KafkaHandleBase(move(config)), message_payload_policy_(PayloadPolicy::COPY_PAYLOAD) {
+: KafkaHandleBase(move(config)), message_payload_policy_(PayloadPolicy::COPY_PAYLOAD),
+  has_internal_data_(false) {
     char error_buffer[512];
     auto config_handle = get_configuration().get_handle();
     rd_kafka_conf_set_opaque(config_handle, this);
@@ -67,47 +69,37 @@ void Producer::produce(const MessageBuilder& builder) {
     const Buffer& payload = builder.payload();
     const Buffer& key = builder.key();
     const int policy = static_cast<int>(message_payload_policy_);
-    void* opaque = builder.user_data();
-    unique_ptr<MessageInternal> internal_data;
-    if (get_configuration().get_delivery_report_callback()) {
-        internal_data.reset(new MessageInternal(builder.user_data(), builder.internal()));
-        opaque = internal_data.get();
-    }
+    LoadResult load_result = load_internal(builder.user_data(), builder.internal());
     auto result = rd_kafka_producev(get_handle(),
                                     RD_KAFKA_V_TOPIC(builder.topic().data()),
                                     RD_KAFKA_V_PARTITION(builder.partition()),
                                     RD_KAFKA_V_MSGFLAGS(policy),
                                     RD_KAFKA_V_TIMESTAMP(builder.timestamp().count()),
                                     RD_KAFKA_V_KEY((void*)key.get_data(), key.get_size()),
                                     RD_KAFKA_V_VALUE((void*)payload.get_data(), payload.get_size()),
-                                    RD_KAFKA_V_OPAQUE(opaque),
+                                    RD_KAFKA_V_OPAQUE(get<0>(load_result)),
                                     RD_KAFKA_V_END);
     check_error(result);
-    internal_data.release(); //data has been passed-on to rdkafka so we release ownership
+    get<1>(load_result).release(); //data has been passed-on to rdkafka so we release ownership
 }
 
 void Producer::produce(const Message& message) {
     const Buffer& payload = message.get_payload();
     const Buffer& key = message.get_key();
     const int policy = static_cast<int>(message_payload_policy_);
     int64_t duration = message.get_timestamp() ? message.get_timestamp().get().get_timestamp().count() : 0;
-    void* opaque = message.get_user_data();
-    unique_ptr<MessageInternal> internal_data;
-    if (get_configuration().get_delivery_report_callback()) {
-        internal_data.reset(new MessageInternal(message.get_user_data(), message.internal()));
-        opaque = internal_data.get();
-    }
+    LoadResult load_result = load_internal(message.get_user_data(), message.internal());
     auto result = rd_kafka_producev(get_handle(),
                                     RD_KAFKA_V_TOPIC(message.get_topic().data()),
                                     RD_KAFKA_V_PARTITION(message.get_partition()),
                                     RD_KAFKA_V_MSGFLAGS(policy),
                                     RD_KAFKA_V_TIMESTAMP(duration),
                                     RD_KAFKA_V_KEY((void*)key.get_data(), key.get_size()),
                                     RD_KAFKA_V_VALUE((void*)payload.get_data(), payload.get_size()),
-                                    RD_KAFKA_V_OPAQUE(opaque),
+                                    RD_KAFKA_V_OPAQUE(get<0>(load_result)),
                                     RD_KAFKA_V_END);
     check_error(result);
-    internal_data.release(); //data has been passed-on to rdkafka so we release ownership
+    get<1>(load_result).release(); //data has been passed-on to rdkafka so we release ownership
 }
 
 int Producer::poll() {
@@ -127,4 +119,16 @@ void Producer::flush(milliseconds timeout) {
     check_error(result);
 }
 
+Producer::LoadResult Producer::load_internal(void* user_data, InternalPtr internal) {
+    unique_ptr<MessageInternal> internal_data;
+    if (!has_internal_data_ && internal) {
+        has_internal_data_ = true; //enable once for this producer
+    }
+    if (has_internal_data_ && get_configuration().get_delivery_report_callback()) {
+        internal_data.reset(new MessageInternal(user_data, internal));
+        user_data = internal_data.get(); //point to the internal data
+    }
+    return LoadResult(user_data, move(internal_data));
+}
+
 } // cppkafka