Allow setting timestamp on produced messages

include/cppkafka/message_builder.h

@@ -30,6 +30,7 @@
 #ifndef CPPKAFKA_MESSAGE_BUILDER_H
 #define CPPKAFKA_MESSAGE_BUILDER_H
 
+#include <chrono>
 #include "buffer.h"
 #include "topic.h"
 #include "macros.h"
@@ -49,6 +50,21 @@ public:
      */
     BasicMessageBuilder(std::string topic);
 
+    /**
+     * \brief Construct a message builder from another one that uses a different buffer type
+     *
+     * Note that this can only be used if BufferType can be constructed from an OtherBufferType
+     *
+     * \param rhs The message builder to be constructed from
+     */
+    template <typename OtherBufferType, typename OtherConcrete>
+    BasicMessageBuilder(const BasicMessageBuilder<OtherBufferType, OtherConcrete>& rhs);
+
+    BasicMessageBuilder(BasicMessageBuilder&&) = default;
+    BasicMessageBuilder(const BasicMessageBuilder&) = default;
+    BasicMessageBuilder& operator=(BasicMessageBuilder&&) = default;
+    BasicMessageBuilder& operator=(const BasicMessageBuilder&) = default;
+
     /**
      * Sets the topic in which this message will be produced
      *
@@ -91,6 +107,13 @@ public:
      */
     Concrete& payload(BufferType&& value);
 
+    /**
+     * Sets the message's timestamp
+     *
+     * \param value The timestamp to be used
+     */
+    Concrete& timestamp(std::chrono::milliseconds value);
+
     /**
      * Sets the message's user data pointer
      *
@@ -128,17 +151,24 @@ public:
      */
     BufferType& payload();
 
+    /**
+     * Gets the message's timestamp
+     */
+    std::chrono::milliseconds timestamp() const;
+
     /**
      * Gets the message's user data pointer
      */
     void* user_data() const;
 private:
     void construct_buffer(BufferType& lhs, const BufferType& rhs);
+    Concrete& get_concrete();
 
     std::string topic_;
     int partition_{-1};
     BufferType key_;
     BufferType payload_;
+    std::chrono::milliseconds timestamp_{0};
     void* user_data_;
 };
 
@@ -147,46 +177,61 @@ BasicMessageBuilder<T, C>::BasicMessageBuilder(std::string topic)
 : topic_(std::move(topic)) {
 }
 
+template <typename T, typename C>
+template <typename U, typename V>
+BasicMessageBuilder<T, C>::BasicMessageBuilder(const BasicMessageBuilder<U, V>& rhs)
+: topic_(rhs.topic()), partition_(rhs.partition()), timestamp_(rhs.timestamp()),
+  user_data_(rhs.user_data()) {
+    get_concrete().construct_buffer(key_, rhs.key());
+    get_concrete().construct_buffer(payload_, rhs.payload());
+}
+
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::topic(std::string value) {
     topic_ = std::move(value);
-    return static_cast<C&>(*this);
+    return get_concrete();
 }
 
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::partition(int value) {
     partition_ = value;
-    return static_cast<C&>(*this);
+    return get_concrete();
 }
 
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::key(const T& value) {
-    static_cast<C&>(*this).construct_buffer(key_, value);
+    get_concrete().construct_buffer(key_, value);
-    return static_cast<C&>(*this);
+    return get_concrete();
 }
 
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::key(T&& value) {
     key_ = std::move(value);
-    return static_cast<C&>(*this);
+    return get_concrete();
 }
 
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::payload(const T& value) {
-    static_cast<C&>(*this).construct_buffer(payload_, value);
+    get_concrete().construct_buffer(payload_, value);
-    return static_cast<C&>(*this);
+    return get_concrete();
 }
 
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::payload(T&& value) {
     payload_ = std::move(value);
-    return static_cast<C&>(*this);
+    return get_concrete();
+}
+
+template <typename T, typename C>
+C& BasicMessageBuilder<T, C>::timestamp(std::chrono::milliseconds value) {
+    timestamp_ = value;
+    return get_concrete();
 }
 
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::user_data(void* value) {
     user_data_ = value;
-    return static_cast<C&>(*this);
+    return get_concrete();
 }
 
 template <typename T, typename C>
@@ -219,6 +264,11 @@ T& BasicMessageBuilder<T, C>::payload() {
     return payload_;
 }
 
+template <typename T, typename C>
+std::chrono::milliseconds BasicMessageBuilder<T, C>::timestamp() const {
+    return timestamp_;
+}
+
 template <typename T, typename C>
 void* BasicMessageBuilder<T, C>::user_data() const {
     return user_data_;
@@ -229,6 +279,11 @@ void BasicMessageBuilder<T, C>::construct_buffer(T& lhs, const T& rhs) {
     lhs = rhs;
 }
 
+template <typename T, typename C>
+C& BasicMessageBuilder<T, C>::get_concrete() {
+    return static_cast<C&>(*this);
+}
+
 /**
  * \brief Message builder class
  * 
@@ -250,6 +305,11 @@ public:
     void construct_buffer(Buffer& lhs, const Buffer& rhs) {
         lhs = Buffer(rhs.get_data(), rhs.get_size());
     }
+
+    template <typename T>
+    void construct_buffer(Buffer& lhs, const T& rhs) {
+        lhs = Buffer(rhs);
+    }
 };
 
 /**

include/cppkafka/utils/buffered_producer.h

@@ -124,13 +124,8 @@ void BufferedProducer<BufferType>::flush() {
 template <typename BufferType>
 template <typename BuilderType>
 void BufferedProducer<BufferType>::do_add_message(BuilderType&& builder) {
-    Builder local_builder(builder.topic());
-    local_builder.partition(builder.partition())
-                 .key(std::move(builder.key()))
-                 .payload(std::move(builder.payload()));
-
     IndexType index = messages_.size();
-    messages_.emplace(index, std::move(local_builder));
+    messages_.emplace(index, std::move(builder));
 }
 
 template <typename BufferType>
@@ -152,11 +147,8 @@ BufferedProducer<BufferType>::make_builder(std::string topic) {
 template <typename BufferType>
 void BufferedProducer<BufferType>::produce_message(IndexType index, Builder& builder) {
     bool sent = false;
-    MessageBuilder local_builder(builder.topic());
+    MessageBuilder local_builder = builder;
-    local_builder.partition(builder.partition())
+    local_builder.user_data(reinterpret_cast<void*>(index));
-                 .key(builder.key())
-                 .payload(builder.payload())
-                 .user_data(reinterpret_cast<void*>(index));
     while (!sent) {
         try {
             producer_.produce(local_builder);

src/producer.cpp

@@ -70,8 +70,9 @@ void Producer::produce(const MessageBuilder& builder) {
                                     RD_KAFKA_V_TOPIC(builder.topic().data()),
                                     RD_KAFKA_V_PARTITION(builder.partition()),
                                     RD_KAFKA_V_MSGFLAGS(policy),
-                                    RD_KAFKA_V_VALUE((void*)payload.get_data(), payload.get_size()),
+                                    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(builder.user_data()),
                                     RD_KAFKA_V_END);
     check_error(result);