Header support for C++ API by @davidtrihy (Superceeds #1959)

examples/rdkafka_example.cpp

@@ -203,6 +203,8 @@ class MyHashPartitionerCb : public RdKafka::PartitionerCb {
 };
 
 void msg_consume(RdKafka::Message* message, void* opaque) {
+  const RdKafka::Headers *headers;
+
   switch (message->err()) {
     case RdKafka::ERR__TIMED_OUT:
       break;
@@ -213,6 +215,20 @@ void msg_consume(RdKafka::Message* message, void* opaque) {
       if (message->key()) {
         std::cout << "Key: " << *message->key() << std::endl;
       }
+      headers = message->headers();
+      if (headers) {
+        std::vector<RdKafka::Headers::Header> hdrs = headers->get_all();
+        for (size_t i = 0 ; i < hdrs.size() ; i++) {
+          const RdKafka::Headers::Header hdr = hdrs[i];
+
+          if (hdr.value() != NULL)
+            printf(" Header: %s = \"%.*s\"\n",
+                   hdr.key().c_str(),
+                   (int)hdr.value_size(), (const char *)hdr.value());
+          else
+            printf(" Header:  %s = NULL\n", hdr.key().c_str());
+        }
+      }
       printf("%.*s\n",
         static_cast<int>(message->len()),
         static_cast<const char *>(message->payload()));
@@ -479,6 +495,8 @@ int main (int argc, char **argv) {
     /* Set delivery report callback */
     conf->set("dr_cb", &ex_dr_cb, errstr);
 
+    conf->set("default_topic_conf", tconf, errstr);
+
     /*
      * Create producer using accumulated global configuration.
      */
@@ -490,15 +508,6 @@ int main (int argc, char **argv) {
 
     std::cout << "% Created producer " << producer->name() << std::endl;
 
-    /*
-     * Create topic handle.
-     */
-    RdKafka::Topic *topic = RdKafka::Topic::create(producer, topic_str,
-						   tconf, errstr);
-    if (!topic) {
-      std::cerr << "Failed to create topic: " << errstr << std::endl;
-      exit(1);
-    }
 
     /*
      * Read messages from stdin and produce to broker.
@@ -509,20 +518,36 @@ int main (int argc, char **argv) {
 	continue;
       }
 
+      RdKafka::Headers *headers = RdKafka::Headers::create();
+      headers->add("my header", "header value");
+      headers->add("other header", "yes");
+
       /*
        * Produce message
        */
       RdKafka::ErrorCode resp =
-	producer->produce(topic, partition,
-			  RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
-			  const_cast<char *>(line.c_str()), line.size(),
-			  NULL, NULL);
-      if (resp != RdKafka::ERR_NO_ERROR)
-	std::cerr << "% Produce failed: " <<
-	  RdKafka::err2str(resp) << std::endl;
-      else
-	std::cerr << "% Produced message (" << line.size() << " bytes)" <<
-	  std::endl;
+        producer->produce(topic_str, partition,
+                          RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
+                          /* Value */
+                          const_cast<char *>(line.c_str()), line.size(),
+                          /* Key */
+                          NULL, 0,
+                          /* Timestamp (defaults to now) */
+                          0,
+                          /* Message headers, if any */
+                          headers,
+                          /* Per-message opaque value passed to
+                           * delivery report */
+                          NULL);
+      if (resp != RdKafka::ERR_NO_ERROR) {
+        std::cerr << "% Produce failed: " <<
+          RdKafka::err2str(resp) << std::endl;
+        delete headers; /* Headers are automatically deleted on produce()
+                         * success. */
+      } else {
+        std::cerr << "% Produced message (" << line.size() << " bytes)" <<
+          std::endl;
+      }
 
       producer->poll(0);
     }
@@ -533,7 +558,6 @@ int main (int argc, char **argv) {
       producer->poll(1000);
     }
 
-    delete topic;
     delete producer;
 
 
@@ -635,7 +659,7 @@ int main (int argc, char **argv) {
       RdKafka::ErrorCode err = producer->metadata(topic!=NULL, topic,
                               &metadata, 5000);
       if (err != RdKafka::ERR_NO_ERROR) {
-        std::cerr << "%% Failed to acquire metadata: " 
+        std::cerr << "%% Failed to acquire metadata: "
                   << RdKafka::err2str(err) << std::endl;
               run = 0;
               break;
@@ -649,6 +673,8 @@ int main (int argc, char **argv) {
 
   }
 
+  delete conf;
+  delete tconf;
 
   /*
    * Wait for RdKafka to decommission.

src-cpp/CMakeLists.txt

@@ -5,6 +5,7 @@ set(
     ConfImpl.cpp
     ConsumerImpl.cpp
     HandleImpl.cpp
+    HeadersImpl.cpp
     KafkaConsumerImpl.cpp
     MessageImpl.cpp
     MetadataImpl.cpp

src-cpp/HeadersImpl.cpp

@@ -0,0 +1,47 @@
+/*
+ * librdkafka - Apache Kafka C/C++ library
+ *
+ * Copyright (c) 2014 Magnus Edenhill
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ * 2. 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 <iostream>
+#include <string>
+#include <list>
+#include <cerrno>
+
+#include "rdkafkacpp_int.h"
+
+RdKafka::Headers *RdKafka::Headers::create() {
+    return new RdKafka::HeadersImpl();
+}
+
+RdKafka::Headers *RdKafka::Headers::create(const std::vector<Header> &headers) {
+  if (headers.size() > 0)
+    return new RdKafka::HeadersImpl(headers);
+  else
+    return new RdKafka::HeadersImpl();
+}
+
+RdKafka::Headers::~Headers() {}

src-cpp/Makefile

@@ -5,7 +5,7 @@ LIBVER=		1
 CXXSRCS=	RdKafka.cpp ConfImpl.cpp HandleImpl.cpp \
 		ConsumerImpl.cpp ProducerImpl.cpp KafkaConsumerImpl.cpp \
 		TopicImpl.cpp TopicPartitionImpl.cpp MessageImpl.cpp \
-		QueueImpl.cpp MetadataImpl.cpp
+		HeadersImpl.cpp QueueImpl.cpp MetadataImpl.cpp
 
 HDRS=		rdkafkacpp.h
 

src-cpp/ProducerImpl.cpp

@@ -143,25 +143,60 @@ RdKafka::ProducerImpl::produce (RdKafka::Topic *topic,
 
 }
 
+RdKafka::ErrorCode
+RdKafka::ProducerImpl::produce (const std::string topic_name,
+                                int32_t partition, int msgflags,
+                                void *payload, size_t len,
+                                const void *key, size_t key_len,
+                                int64_t timestamp, void *msg_opaque) {
+  return
+  static_cast<RdKafka::ErrorCode>
+    (
+      rd_kafka_producev(rk_,
+                        RD_KAFKA_V_TOPIC(topic_name.c_str()),
+                        RD_KAFKA_V_PARTITION(partition),
+                        RD_KAFKA_V_MSGFLAGS(msgflags),
+                        RD_KAFKA_V_VALUE(payload, len),
+                        RD_KAFKA_V_KEY(key, key_len),
+                        RD_KAFKA_V_TIMESTAMP(timestamp),
+                        RD_KAFKA_V_OPAQUE(msg_opaque),
+                        RD_KAFKA_V_END)
+      );
+}
 
 RdKafka::ErrorCode
 RdKafka::ProducerImpl::produce (const std::string topic_name,
                                 int32_t partition, int msgflags,
                                 void *payload, size_t len,
                                 const void *key, size_t key_len,
                                 int64_t timestamp,
+                                RdKafka::Headers *headers,
                                 void *msg_opaque) {
-  return
-    static_cast<RdKafka::ErrorCode>
-    (
-     rd_kafka_producev(rk_,
-                       RD_KAFKA_V_TOPIC(topic_name.c_str()),
-                       RD_KAFKA_V_PARTITION(partition),
-                       RD_KAFKA_V_MSGFLAGS(msgflags),
-                       RD_KAFKA_V_VALUE(payload, len),
-                       RD_KAFKA_V_KEY(key, key_len),
-                       RD_KAFKA_V_TIMESTAMP(timestamp),
-                       RD_KAFKA_V_OPAQUE(msg_opaque),
-                       RD_KAFKA_V_END)
-     );
+  rd_kafka_headers_t *hdrs = NULL;
+  RdKafka::HeadersImpl *headersimpl = NULL;
+  rd_kafka_resp_err_t err;
+
+  if (headers) {
+    headersimpl = static_cast<RdKafka::HeadersImpl*>(headers);
+    hdrs = headersimpl->c_ptr();
+  }
+
+  err = rd_kafka_producev(rk_,
+                          RD_KAFKA_V_TOPIC(topic_name.c_str()),
+                          RD_KAFKA_V_PARTITION(partition),
+                          RD_KAFKA_V_MSGFLAGS(msgflags),
+                          RD_KAFKA_V_VALUE(payload, len),
+                          RD_KAFKA_V_KEY(key, key_len),
+                          RD_KAFKA_V_TIMESTAMP(timestamp),
+                          RD_KAFKA_V_OPAQUE(msg_opaque),
+                          RD_KAFKA_V_HEADERS(hdrs),
+                          RD_KAFKA_V_END);
+
+  if (!err && headersimpl) {
+    /* A successful producev() call will destroy the C headers. */
+    headersimpl->c_headers_destroyed();
+    delete headers;
+  }
+
+  return static_cast<RdKafka::ErrorCode>(err);
 }