Added time_point overloads for creating timestamps.

Author: accelerated

Diff for: include/cppkafka/message.h

@@ -240,12 +240,19 @@ class CPPKAFKA_API MessageTimestamp {
     };
 
     /**
-     * Constructs a timestamp object
+     * Constructs a timestamp object using a 'duration'.
      */
     MessageTimestamp(std::chrono::milliseconds timestamp, TimestampType type);
+    
+    /**
+     * Constructs a timestamp object using a 'time_point'.
+     */
+    template <typename Clock, typename Duration = typename Clock::duration>
+    MessageTimestamp(std::chrono::time_point<Clock, Duration> timestamp, TimestampType type);
 
     /**
-     * Gets the timestamp value
+     * Gets the timestamp value. If the timestamp was created with a 'time_point',
+     * the duration represents the number of milliseconds since epoch.
      */
     std::chrono::milliseconds get_timestamp() const;
 

Diff for: include/cppkafka/message_builder.h

@@ -128,11 +128,19 @@ class BasicMessageBuilder {
     Concrete& payload(BufferType&& value);
 
     /**
-     * Sets the message's timestamp
+     * Sets the message's timestamp with a 'duration'
      *
      * \param value The timestamp to be used
      */
     Concrete& timestamp(std::chrono::milliseconds value);
+    
+    /**
+     * Sets the message's timestamp with a 'time_point'.
+     *
+     * \param value The timestamp to be used
+     */
+    template <typename Clock, typename Duration = typename Clock::duration>
+    Concrete& timestamp(std::chrono::time_point<Clock, Duration> value);
 
     /**
      * Sets the message's user data pointer
@@ -184,7 +192,8 @@ class BasicMessageBuilder {
     BufferType& payload();
 
     /**
-     * Gets the message's timestamp
+     * Gets the message's timestamp as a duration. If the timestamp was created with a 'time_point',
+     * the duration represents the number of milliseconds since epoch.
      */
     std::chrono::milliseconds timestamp() const;
 
@@ -295,6 +304,14 @@ C& BasicMessageBuilder<T, C>::timestamp(std::chrono::milliseconds value) {
     return get_concrete();
 }
 
+template <typename T, typename C>
+template <typename Clock, typename Duration>
+C& BasicMessageBuilder<T, C>::timestamp(std::chrono::time_point<Clock, Duration> value)
+{
+    timestamp_ = std::chrono::duration_cast<std::chrono::milliseconds>(value.time_since_epoch());
+    return get_concrete();
+}
+
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::user_data(void* value) {
     user_data_ = value;

Diff for: src/message.cpp

@@ -87,10 +87,18 @@ Message& Message::load_internal() {
 // MessageTimestamp
 
 MessageTimestamp::MessageTimestamp(milliseconds timestamp, TimestampType type)
-: timestamp_(timestamp), type_(type) {
+: timestamp_(timestamp),
+  type_(type) {
 
 }
 
+template <typename Clock, typename Duration>
+MessageTimestamp::MessageTimestamp(std::chrono::time_point<Clock, Duration> timestamp, TimestampType type)
+: timestamp_(std::chrono::duration_cast<std::chrono::milliseconds>(timestamp.time_since_epoch())),
+  type_(type) {
+  
+}
+
 milliseconds MessageTimestamp::get_timestamp() const {
     return timestamp_;
 }

Diff for: tests/producer_test.cpp

@@ -164,7 +164,7 @@ TEST_CASE("simple production", "[producer]") {
     SECTION("message with key") {
         const string payload = "Hello world! 2";
         const string key = "such key";
-        const milliseconds timestamp{15};
+        auto timestamp = std::chrono::system_clock::now();
         Producer producer(config);
         producer.produce(MessageBuilder(KAFKA_TOPICS[0]).partition(partition)
                                                      .key(key)
@@ -181,7 +181,7 @@ TEST_CASE("simple production", "[producer]") {
         CHECK(message.get_partition() == partition);
         CHECK(!!message.get_error() == false);
         REQUIRE(!!message.get_timestamp() == true);
-        CHECK(message.get_timestamp()->get_timestamp() == timestamp);
+        CHECK(message.get_timestamp()->get_timestamp() == std::chrono::duration_cast<milliseconds>(timestamp.time_since_epoch()));
     }
 
 #if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)