Header support implementation (#115)

* header support implementation

* Fixed issue when ptr is null and doesn't have a cloner function

* Code complete with test cases

updated travis file with v0.11.5

* Added compile time check for rdkafka header support version

* Changes per last code review

* Using brace list initializers

include/cppkafka/buffer.h

@@ -172,6 +172,14 @@ CPPKAFKA_API bool operator==(const Buffer& lhs, const Buffer& rhs);
  */
 CPPKAFKA_API bool operator!=(const Buffer& lhs, const Buffer& rhs);
 
+/**
+ * Compares Buffer objects lexicographically
+ */
+CPPKAFKA_API bool operator<(const Buffer& lhs, const Buffer& rhs);
+CPPKAFKA_API bool operator<=(const Buffer& lhs, const Buffer& rhs);
+CPPKAFKA_API bool operator>(const Buffer& lhs, const Buffer& rhs);
+CPPKAFKA_API bool operator>=(const Buffer& lhs, const Buffer& rhs);
+
 } // cppkafka
 
 #endif // CPPKAFKA_BUFFER_H

include/cppkafka/clonable_ptr.h

@@ -41,7 +41,7 @@ template <typename T, typename Deleter, typename Cloner>
 class ClonablePtr {
 public:
     /**
-     * Creates an instance
+     * \brief Creates an instance
      *
      * \param ptr The pointer to be wrapped
      * \param deleter The deleter functor
@@ -60,17 +60,23 @@ public:
      * \param rhs The pointer to be copied
      */
     ClonablePtr(const ClonablePtr& rhs)
-    : handle_(rhs.cloner_(rhs.handle_.get()), rhs.handle_.get_deleter()), cloner_(rhs.cloner_) {
+    : handle_(rhs.cloner_ ? std::unique_ptr<T, Deleter>(rhs.cloner_(rhs.handle_.get()), rhs.handle_.get_deleter()) :
+                            std::unique_ptr<T, Deleter>(rhs.handle_.get(), rhs.handle_.get_deleter())),
+      cloner_(rhs.cloner_) {
 
     }
 
     /** 
-     * Copies and assigns the given pointer
+     * \brief Copies and assigns the given pointer
      *
      * \param rhs The pointer to be copied
      */
     ClonablePtr& operator=(const ClonablePtr& rhs) {
-        handle_.reset(cloner_(rhs.handle_.get()));
+        if (this == &rhs) {
+            return *this;
+        }
+        handle_ = rhs.cloner_ ? std::unique_ptr<T, Deleter>(rhs.cloner_(rhs.handle_.get()), rhs.handle_.get_deleter()) :
+                                std::unique_ptr<T, Deleter>(rhs.handle_.get(), rhs.handle_.get_deleter());
         return *this;
     }
 
@@ -79,11 +85,25 @@ public:
     ~ClonablePtr() = default;
 
     /**
-     * Getter for the internal pointer
+     * \brief Getter for the internal pointer
      */
     T* get() const {
         return handle_.get();
     }
+    
+    /**
+     * \brief Releases ownership of the internal pointer
+     */
+    T* release() {
+        return handle_.release();
+    }
+    
+    /**
+     * \brief Indicates whether this ClonablePtr instance is valid (not null)
+     */
+    explicit operator bool() const {
+        return static_cast<bool>(handle_);
+    }
 private:
     std::unique_ptr<T, Deleter> handle_;
     Cloner cloner_;

include/cppkafka/cppkafka.h

@@ -39,6 +39,9 @@
 #include <cppkafka/error.h>
 #include <cppkafka/exceptions.h>
 #include <cppkafka/group_information.h>
+#include <cppkafka/header.h>
+#include <cppkafka/header_list.h>
+#include <cppkafka/header_list_iterator.h>
 #include <cppkafka/kafka_handle_base.h>
 #include <cppkafka/logging.h>
 #include <cppkafka/macros.h>

include/cppkafka/header.h

@@ -0,0 +1,195 @@
+/*
+ * 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.
+ *
+ */
+ 
+#ifndef CPPKAFKA_HEADER_H
+#define CPPKAFKA_HEADER_H
+
+#include "macros.h"
+#include "buffer.h"
+#include <string>
+#include <assert.h>
+
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+
+namespace cppkafka {
+
+/**
+ * \brief Class representing a rdkafka header.
+ *
+ * The template parameter 'BufferType' can represent a cppkafka::Buffer, std::string, std::vector, etc.
+ * A valid header may contain an empty name as well as null data.
+ */
+template <typename BufferType>
+class Header {
+public:
+    using ValueType = BufferType;
+    
+    /**
+     * \brief Build an empty header with no data
+     */
+    Header() = default;
+    
+    /**
+     * \brief Build a header instance
+     * \param name The header name
+     * \param value The non-modifiable header data
+     */
+    Header(std::string name,
+           const BufferType& value);
+    
+    /**
+     * \brief Build a header instance
+     * \param name The header name
+     * \param value The header data to be moved
+     */
+    Header(std::string name,
+           BufferType&& value);
+    
+    /**
+     * \brief Get the header name
+     * \return A reference to the name
+     */
+    const std::string& get_name() const;
+    
+    /**
+     * \brief Get the header value
+     * \return A const reference to the underlying buffer
+     */
+    const BufferType& get_value() const;
+    
+    /**
+     * \brief Get the header value
+     * \return A non-const reference to the underlying buffer
+     */
+    BufferType& get_value();
+    
+    /**
+     * \brief Check if this header is empty
+     * \return True if the header contains valid data, false otherwise.
+     */
+    operator bool() const;
+    
+private:
+    template <typename T>
+    T make_value(const T& other);
+    
+    Buffer make_value(const Buffer& other);
+    
+    std::string name_;
+    BufferType value_;
+};
+
+// Comparison operators for Header type
+template <typename BufferType>
+bool operator==(const Header<BufferType>& lhs, const Header<BufferType>& rhs) {
+    return std::tie(lhs.get_name(), lhs.get_value()) == std::tie(rhs.get_name(), rhs.get_value());
+}
+
+template <typename BufferType>
+bool operator!=(const Header<BufferType>& lhs, const Header<BufferType>& rhs) {
+    return !(lhs == rhs);
+}
+
+template <typename BufferType>
+bool operator<(const Header<BufferType>& lhs, const Header<BufferType>& rhs) {
+    return std::tie(lhs.get_name(), lhs.get_value()) < std::tie(rhs.get_name(), rhs.get_value());
+}
+
+template <typename BufferType>
+bool operator>(const Header<BufferType>& lhs, const Header<BufferType>& rhs) {
+    return std::tie(lhs.get_name(), lhs.get_value()) > std::tie(rhs.get_name(), rhs.get_value());
+}
+
+template <typename BufferType>
+bool operator<=(const Header<BufferType>& lhs, const Header<BufferType>& rhs) {
+    return !(lhs > rhs);
+}
+
+template <typename BufferType>
+bool operator>=(const Header<BufferType>& lhs, const Header<BufferType>& rhs) {
+    return !(lhs < rhs);
+}
+
+// Implementation
+template <typename BufferType>
+Header<BufferType>::Header(std::string name,
+                           const BufferType& value)
+: name_(std::move(name)),
+  value_(make_value(value)) {
+}
+
+template <typename BufferType>
+Header<BufferType>::Header(std::string name,
+                           BufferType&& value)
+: name_(std::move(name)),
+  value_(std::move(value)) {
+}
+
+template <typename BufferType>
+const std::string& Header<BufferType>::get_name() const {
+    return name_;
+}
+
+template <typename BufferType>
+const BufferType& Header<BufferType>::get_value() const {
+    return value_;
+}
+
+template <typename BufferType>
+BufferType& Header<BufferType>::get_value() {
+    return value_;
+}
+
+template <typename BufferType>
+Header<BufferType>::operator bool() const {
+    return !value_.empty();
+}
+
+template <>
+inline
+Header<Buffer>::operator bool() const {
+    return value_.get_size() > 0;
+}
+
+template <typename BufferType>
+template <typename T>
+T Header<BufferType>::make_value(const T& other) {
+    return other;
+}
+
+template <typename BufferType>
+Buffer Header<BufferType>::make_value(const Buffer& other) {
+    return Buffer(other.get_data(), other.get_size());
+}
+
+} //namespace cppkafka
+
+#endif //RD_KAFKA_HEADERS_SUPPORT_VERSION
+
+#endif //CPPKAFKA_HEADER_H

include/cppkafka/header_list.h

@@ -0,0 +1,317 @@
+/*
+ * 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.
+ *
+ */
+ 
+#ifndef CPPKAFKA_HEADER_LIST_H
+#define CPPKAFKA_HEADER_LIST_H
+
+#include <librdkafka/rdkafka.h>
+#include "clonable_ptr.h"
+#include "header.h"
+#include "header_list_iterator.h"
+#include "exceptions.h"
+
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+
+namespace cppkafka {
+
+/**
+ * \brief Thin wrapper over a rd_kafka_headers_t handle which optionally controls its lifetime.
+ * \tparam HeaderType The header type
+ *
+ * This is a copyable and movable class that wraps a rd_kafka_header_t*. When copying this class,
+ * all associated headers are also copied via rd_kafka_headers_copy(). If this list owns the underlying handle,
+ * its destructor will call rd_kafka_headers_destroy().
+ */
+template <typename HeaderType>
+class HeaderList {
+public:
+    using BufferType = typename HeaderType::ValueType;
+    using Iterator = HeaderIterator<HeaderType>;
+    /**
+     * Constructs a message that won't take ownership of the given pointer.
+     */
+    static HeaderList<HeaderType> make_non_owning(rd_kafka_headers_t* handle);
+    
+    /**
+     * \brief Create an empty header list with no handle.
+     */
+    HeaderList();
+    
+    /**
+     * \brief Create an empty header list. This call translates to rd_kafka_headers_new().
+     * \param reserve The number of headers to reserve space for.
+     */
+    explicit HeaderList(size_t reserve);
+    
+    /**
+     * \brief Create a header list and assume ownership of the handle.
+     * \param handle The header list handle.
+     */
+    explicit HeaderList(rd_kafka_headers_t* handle);
+    
+    /**
+     * \brief Add a header to the list. This translates to rd_kafka_header_add().
+     * \param header The header.
+     * \return An Error indicating if the operation was successful or not.
+     * \warning This operation shall invalidate all iterators.
+     */
+    Error add(const HeaderType& header);
+    
+    /**
+     * \brief Remove all headers with 'name'. This translates to rd_kafka_header_remove().
+     * \param name The name of the header(s) to remove.
+     * \return An Error indicating if the operation was successful or not.
+     * \warning This operation shall invalidate all iterators.
+     */
+    Error remove(const std::string& name);
+    
+    /**
+     * \brief Return the header present at position 'index'. Throws on error.
+     *        This translates to rd_kafka_header_get(index)
+     * \param index The header index in the list (0-based).
+     * \return The header at that position.
+     */
+    HeaderType at(size_t index) const; //throws
+    
+    /**
+     * \brief Return the first header in the list. Throws if the list is empty.
+     *        This translates to rd_kafka_header_get(0).
+     * \return The first header.
+     */
+    HeaderType front() const; //throws
+    
+    /**
+     * \brief Return the first header in the list. Throws if the list is empty.
+     *        This translates to rd_kafka_header_get(size-1).
+     * \return The last header.
+     */
+    HeaderType back() const; //throws
+    
+    /**
+     * \brief Returns the number of headers in the list. This translates to rd_kafka_header_cnt().
+     * \return The number of headers.
+     */
+    size_t size() const;
+    
+    /**
+     * \brief Indicates if this list is empty.
+     * \return True if empty, false otherwise.
+     */
+    bool empty() const;
+    
+    /**
+     * \brief Returns a HeaderIterator pointing to the first position if the list is not empty
+     *        or pointing to end() otherwise.
+     * \return An iterator.
+     * \warning This iterator will be invalid if add() or remove() is called.
+     */
+    Iterator begin() const;
+    
+    /**
+     * \brief Returns a HeaderIterator pointing to one element past the end of the list.
+     * \return An iterator.
+     * \remark This iterator cannot be de-referenced.
+     */
+    Iterator end() const;
+    
+    /**
+     * \brief Get the underlying header list handle.
+     * \return The handle.
+     */
+    rd_kafka_headers_t* get_handle() const;
+    
+    /**
+     * \brief Get the underlying header list handle and release its ownership.
+     * \return The handle.
+     * \warning After this call, the HeaderList becomes invalid.
+     */
+    rd_kafka_headers_t* release_handle();
+    
+    /**
+     * \brief Indicates if this list is valid (contains a non-null handle) or not.
+     * \return True if valid, false otherwise.
+     */
+    explicit operator bool() const;
+    
+private:
+    struct NonOwningTag { };
+    static void dummy_deleter(rd_kafka_headers_t*) {}
+    static rd_kafka_headers_t* dummy_cloner(const rd_kafka_headers_t* handle) { return const_cast<rd_kafka_headers_t*>(handle); }
+    
+    using HandlePtr = ClonablePtr<rd_kafka_headers_t, decltype(&rd_kafka_headers_destroy),
+                                  decltype(&rd_kafka_headers_copy)>;
+    
+    HeaderList(rd_kafka_headers_t* handle, NonOwningTag);
+    
+    HandlePtr handle_;
+};
+
+template <typename HeaderType>
+bool operator==(const HeaderList<HeaderType>& lhs, const HeaderList<HeaderType> rhs) {
+    if (!lhs && !rhs) {
+        return true;
+    }
+    if (!lhs || !rhs) {
+        return false;
+    }
+    if (lhs.size() != rhs.size()) {
+        return false;
+    }
+    return std::equal(lhs.begin(), lhs.end(), rhs.begin());
+}
+
+template <typename HeaderType>
+bool operator!=(const HeaderList<HeaderType>& lhs, const HeaderList<HeaderType> rhs) {
+    return !(lhs == rhs);
+}
+
+template <typename HeaderType>
+HeaderList<HeaderType> HeaderList<HeaderType>::make_non_owning(rd_kafka_headers_t* handle) {
+    return HeaderList(handle, NonOwningTag());
+}
+
+template <typename HeaderType>
+HeaderList<HeaderType>::HeaderList()
+: handle_(nullptr, nullptr, nullptr) {
+
+}
+
+template <typename HeaderType>
+HeaderList<HeaderType>::HeaderList(size_t reserve)
+: handle_(rd_kafka_headers_new(reserve), &rd_kafka_headers_destroy, &rd_kafka_headers_copy) {
+
+}
+
+template <typename HeaderType>
+HeaderList<HeaderType>::HeaderList(rd_kafka_headers_t* handle)
+: handle_(handle, &rd_kafka_headers_destroy, &rd_kafka_headers_copy) { //if we own the header list, we clone it on copy
+
+}
+
+template <typename HeaderType>
+HeaderList<HeaderType>::HeaderList(rd_kafka_headers_t* handle, NonOwningTag)
+: handle_(HandlePtr(handle, &dummy_deleter, &dummy_cloner)) { //if we don't own the header list, we forward the handle on copy.
+
+}
+
+// Methods
+template <typename HeaderType>
+Error HeaderList<HeaderType>::add(const HeaderType& header) {
+    assert(handle_);
+    return rd_kafka_header_add(handle_.get(),
+                               header.get_name().data(),  header.get_name().size(),
+                               header.get_value().data(), header.get_value().size());
+    
+}
+
+template <>
+inline
+Error HeaderList<Header<Buffer>>::add(const Header<Buffer>& header) {
+    assert(handle_);
+    return rd_kafka_header_add(handle_.get(),
+                               header.get_name().data(), header.get_name().size(),
+                               header.get_value().get_data(), header.get_value().get_size());
+}
+
+template <typename HeaderType>
+Error HeaderList<HeaderType>::remove(const std::string& name) {
+    assert(handle_);
+    return rd_kafka_header_remove(handle_.get(), name.data());
+}
+
+template <typename HeaderType>
+HeaderType HeaderList<HeaderType>::at(size_t index) const {
+    assert(handle_);
+    const char *name, *value;
+    size_t size;
+    Error error = rd_kafka_header_get_all(handle_.get(), index, &name, reinterpret_cast<const void**>(&value), &size);
+    if (error != RD_KAFKA_RESP_ERR_NO_ERROR) {
+        throw Exception(error.to_string());
+    }
+    return HeaderType(name, BufferType(value, size));
+}
+
+template <typename HeaderType>
+HeaderType HeaderList<HeaderType>::front() const {
+    return at(0);
+}
+
+template <typename HeaderType>
+HeaderType HeaderList<HeaderType>::back() const {
+    return at(size()-1);
+}
+
+template <typename HeaderType>
+size_t HeaderList<HeaderType>::size() const {
+    assert(handle_);
+    return rd_kafka_header_cnt(handle_.get());
+}
+
+template <typename HeaderType>
+bool HeaderList<HeaderType>::empty() const {
+    return size() == 0;
+}
+
+template <typename HeaderType>
+typename HeaderList<HeaderType>::Iterator
+HeaderList<HeaderType>::begin() const {
+    assert(handle_);
+    if (empty()) {
+        return end();
+    }
+    return Iterator(make_non_owning(handle_.get()), 0);
+}
+
+template <typename HeaderType>
+typename HeaderList<HeaderType>::Iterator
+HeaderList<HeaderType>::end() const {
+    assert(handle_);
+    return Iterator(make_non_owning(handle_.get()), size());
+}
+
+template <typename HeaderType>
+rd_kafka_headers_t* HeaderList<HeaderType>::get_handle() const {
+    return handle_.get();
+}
+
+template <typename HeaderType>
+rd_kafka_headers_t* HeaderList<HeaderType>::release_handle() {
+    return handle_.release();
+}
+
+template <typename HeaderType>
+HeaderList<HeaderType>::operator bool() const {
+    return static_cast<bool>(handle_);
+}
+
+} //namespace cppkafka
+
+#endif //RD_KAFKA_HEADERS_SUPPORT_VERSION
+
+#endif //CPPKAFKA_HEADER_LIST_H

include/cppkafka/header_list_iterator.h

@@ -0,0 +1,193 @@
+/*
+ * 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.
+ *
+ */
+ 
+#ifndef CPPKAFKA_HEADER_LIST_ITERATOR_H
+#define CPPKAFKA_HEADER_LIST_ITERATOR_H
+
+#include <cstddef>
+#include <utility>
+#include <iterator>
+#include "header.h"
+
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+
+namespace cppkafka {
+
+template <typename HeaderType>
+class HeaderList;
+
+template <typename HeaderType>
+class HeaderIterator;
+
+template <typename HeaderType>
+bool operator==(const HeaderIterator<HeaderType>& lhs, const HeaderIterator<HeaderType>& rhs);
+
+/**
+ * \brief Iterator over a HeaderList object.
+ * \tparam HeaderType The type of header this iterator points to.
+ */
+template <typename HeaderType>
+class HeaderIterator {
+public:
+    friend HeaderList<HeaderType>;
+    using HeaderListType = HeaderList<HeaderType>;
+    using BufferType = typename HeaderType::ValueType;
+    //std::iterator_traits
+    using difference_type = std::ptrdiff_t;
+    using value_type = HeaderType;
+    using pointer = value_type*;
+    using reference = value_type&;
+    using iterator_category = std::bidirectional_iterator_tag;
+    friend bool operator==<HeaderType>(const HeaderIterator<HeaderType>& lhs,
+                                       const HeaderIterator<HeaderType>& rhs);
+    
+    HeaderIterator(const HeaderIterator& other)
+    : header_list_(other.header_list_),
+      header_(make_header(other.header_)),
+      index_(other.index_) {
+      
+    }
+    HeaderIterator& operator=(const HeaderIterator& other) {
+        if (this == &other) return *this;
+        header_list_ = other.header_list_;
+        header_ = make_header(other.header_);
+        index_ = other.index_;
+    }
+    HeaderIterator(HeaderIterator&&) = default;
+    HeaderIterator& operator=(HeaderIterator&&) = default;
+    
+    /**
+     * \brief Prefix increment of the iterator.
+     * \return Itself after being incremented.
+     */
+    HeaderIterator& operator++() {
+        assert(index_ < header_list_.size());
+        ++index_;
+        return *this;
+    }
+    
+    /**
+     * \brief Postfix increment of the iterator.
+     * \return Itself before being incremented.
+     */
+    HeaderIterator operator++(int) {
+        HeaderIterator tmp(*this);
+        operator++();
+        return tmp;
+    }
+    
+    /**
+     * \brief Prefix decrement of the iterator.
+     * \return Itself after being decremented.
+     */
+    HeaderIterator& operator--() {
+        assert(index_ > 0);
+        --index_;
+        return *this;
+    }
+    
+    /**
+     * \brief Postfix decrement of the iterator.
+     * \return Itself before being decremented.
+     */
+    HeaderIterator operator--(int) {
+        HeaderIterator tmp(*this);
+        operator--();
+        return tmp;
+    }
+    
+    /**
+     * \brief Dereferences this iterator.
+     * \return A reference to the header the iterator points to.
+     * \warning Throws if invalid or if *this == end().
+     */
+    const HeaderType& operator*() const {
+        header_ = header_list_.at(index_);
+        return header_;
+    }
+    HeaderType& operator*() {
+        header_ = header_list_.at(index_);
+        return header_;
+    }
+    
+    /**
+     * \brief Dereferences this iterator.
+     * \return The address to the header the iterator points to.
+     * \warning Throws if invalid or if *this == end().
+     */
+    const HeaderType* operator->() const {
+        header_ = header_list_.at(index_);
+        return &header_;
+    }
+    HeaderType* operator->() {
+        header_ = header_list_.at(index_);
+        return &header_;
+    }
+    
+private:
+    HeaderIterator(HeaderListType headers,
+                   size_t index)
+    : header_list_(std::move(headers)),
+      header_(index == header_list_.size() ? HeaderType() : header_list_.at(index)),
+      index_(index) {
+    }
+    
+    template <typename T>
+    T make_header(const T& other) {
+        return other;
+    }
+    
+    Header<Buffer> make_header(const Header<Buffer>& other) {
+        return Header<Buffer>(other.get_name(),
+                              Buffer(other.get_value().get_data(),
+                                     other.get_value().get_size()));
+    }
+    
+    HeaderListType header_list_;
+    HeaderType header_;
+    size_t index_;
+};
+
+// Equality comparison operators
+template <typename HeaderType>
+bool operator==(const HeaderIterator<HeaderType>& lhs, const HeaderIterator<HeaderType>& rhs) {
+    return (lhs.header_list_.get_handle() == rhs.header_list_.get_handle()) && (lhs.index_ == rhs.index_);
+}
+
+template <typename HeaderType>
+bool operator!=(const HeaderIterator<HeaderType>& lhs, const HeaderIterator<HeaderType>& rhs) {
+    return !(lhs == rhs);
+}
+
+} //namespace cppkafka
+
+#endif //RD_KAFKA_HEADERS_SUPPORT_VERSION
+
+#endif //CPPKAFKA_HEADER_LIST_ITERATOR_H
+

include/cppkafka/macros.h

@@ -43,4 +43,8 @@
     #define CPPKAFKA_API
 #endif // _WIN32 && !CPPKAFKA_STATIC
 
+// See: https://github.com/edenhill/librdkafka/issues/1792
+#define RD_KAFKA_QUEUE_REFCOUNT_BUG_VERSION 0x000b0500 //v0.11.5.00
+#define RD_KAFKA_HEADERS_SUPPORT_VERSION 0x000b0402 //v0.11.4.02
+
 #endif // CPPKAFKA_MACROS_H

include/cppkafka/message.h

@@ -39,6 +39,7 @@
 #include "buffer.h"
 #include "macros.h"
 #include "error.h"
+#include "header_list.h"
 
 namespace cppkafka {
 
@@ -59,6 +60,10 @@ class CPPKAFKA_API Message {
 public:
     friend class MessageInternal;
     using InternalPtr = std::shared_ptr<Internal>;
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    using HeaderType = Header<Buffer>;
+    using HeaderListType = HeaderList<HeaderType>;
+#endif
     /**
      * Constructs a message that won't take ownership of the given pointer
      */
@@ -84,7 +89,7 @@ public:
     Message& operator=(Message&& rhs) = default;
 
     /**
-     * Gets the error attribute
+     * \brief Gets the error attribute
      */
     Error get_error() const {
         assert(handle_);
@@ -92,14 +97,14 @@ public:
     }
 
     /**
-     * Utility function to check for get_error() == RD_KAFKA_RESP_ERR__PARTITION_EOF
+     * \brief Utility function to check for get_error() == RD_KAFKA_RESP_ERR__PARTITION_EOF
      */
     bool is_eof() const {
         return get_error() == RD_KAFKA_RESP_ERR__PARTITION_EOF;
     }
 
     /**
-     * Gets the topic that this message belongs to
+     * \brief Gets the topic that this message belongs to
      */
     std::string get_topic() const {
         assert(handle_);
@@ -107,7 +112,7 @@ public:
     }
 
     /**
-     * Gets the partition that this message belongs to
+     * \brief Gets the partition that this message belongs to
      */
     int get_partition() const {
         assert(handle_);
@@ -115,21 +120,40 @@ public:
     }
 
     /**
-     * Gets the message's payload
+     * \brief Gets the message's payload
      */
     const Buffer& get_payload() const {
         return payload_;
     }
+    
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    /**
+     * \brief Gets the message's header list
+     */
+    const HeaderListType& get_header_list() const {
+        return header_list_;
+    }
+    
+    /**
+     * \brief Detaches the message's header list
+     */
+    template <typename HeaderType>
+    HeaderList<HeaderType> detach_header_list() {
+        rd_kafka_headers_t* headers_handle;
+        Error error = rd_kafka_message_detach_headers(handle_.get(), &headers_handle);
+        return error ? HeaderList<HeaderType>() : HeaderList<HeaderType>(headers_handle);
+    }
+#endif
 
     /**
-     * Gets the message's key
+     * \brief Gets the message's key
      */
     const Buffer& get_key() const {
         return key_;
     }
 
     /**
-     * Gets the message offset
+     * \brief Gets the message offset
      */
     int64_t get_offset() const {
         assert(handle_);
@@ -152,23 +176,31 @@ public:
      * If calling rd_kafka_message_timestamp returns -1, then boost::none_t will be returned.
      */
     inline boost::optional<MessageTimestamp> get_timestamp() const;
+    
+    /**
+     * \brief Gets the message latency in microseconds as measured from the produce() call.
+     */
+    std::chrono::microseconds get_latency() const {
+        assert(handle_);
+        return std::chrono::microseconds(rd_kafka_message_latency(handle_.get()));
+    }
 
     /**
-     * Indicates whether this message is valid (not null)
+     * \brief Indicates whether this message is valid (not null)
      */
     explicit operator bool() const {
         return handle_ != nullptr;
     }
 
     /**
-     * Gets the rdkafka message handle
+     * \brief Gets the rdkafka message handle
      */
     rd_kafka_message_t* get_handle() const {
         return handle_.get();
     }
     
     /**
-     * Internal private const data accessor (internal use only)
+     * \brief Internal private const data accessor (internal use only)
      */
     InternalPtr internal() const {
         return internal_;
@@ -185,6 +217,9 @@ private:
     HandlePtr handle_;
     Buffer payload_;
     Buffer key_;
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    HeaderListType header_list_;
+#endif
     void* user_data_;
     InternalPtr internal_;
 };

include/cppkafka/message_builder.h

@@ -35,6 +35,7 @@
 #include "topic.h"
 #include "macros.h"
 #include "message.h"
+#include "header_list.h"
 
 namespace cppkafka {
 
@@ -44,6 +45,10 @@ namespace cppkafka {
 template <typename BufferType, typename Concrete>
 class BasicMessageBuilder {
 public:
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    using HeaderType = Header<BufferType>;
+    using HeaderListType = HeaderList<HeaderType>;
+#endif
     /**
      * Construct a BasicMessageBuilder
      *
@@ -99,6 +104,15 @@ public:
      */
     Concrete& key(BufferType&& value);
 
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    /**
+     *  Add a header to the message
+     *
+     * \param header The header to be used
+     */
+    Concrete& header(const HeaderType& header);
+#endif
+    
     /**
      * Sets the message's payload
      *
@@ -146,7 +160,19 @@ public:
      * Gets the message's key
      */
     BufferType& key();
-
+    
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    /**
+     * Gets the list of headers
+     */
+    const HeaderListType& header_list() const;
+    
+    /**
+     * Gets the list of headers
+     */
+    HeaderListType& header_list();
+#endif
+    
     /**
      * Gets the message's payload
      */
@@ -180,6 +206,9 @@ private:
     std::string topic_;
     int partition_{-1};
     BufferType key_;
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    HeaderListType header_list_;
+#endif
     BufferType payload_;
     std::chrono::milliseconds timestamp_{0};
     void* user_data_;
@@ -237,6 +266,17 @@ C& BasicMessageBuilder<T, C>::key(T&& value) {
     return get_concrete();
 }
 
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+template <typename T, typename C>
+C& BasicMessageBuilder<T, C>::header(const HeaderType& header) {
+    if (!header_list_) {
+        header_list_ = HeaderListType(5);
+    }
+    header_list_.add(header);
+    return get_concrete();
+}
+#endif
+
 template <typename T, typename C>
 C& BasicMessageBuilder<T, C>::payload(const T& value) {
     get_concrete().construct_buffer(payload_, value);
@@ -281,6 +321,20 @@ T& BasicMessageBuilder<T, C>::key() {
     return key_;
 }
 
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+template <typename T, typename C>
+const typename BasicMessageBuilder<T, C>::HeaderListType&
+BasicMessageBuilder<T, C>::header_list() const {
+    return header_list_;
+}
+
+template <typename T, typename C>
+typename BasicMessageBuilder<T, C>::HeaderListType&
+BasicMessageBuilder<T, C>::header_list() {
+    return header_list_;
+}
+#endif
+
 template <typename T, typename C>
 const T& BasicMessageBuilder<T, C>::payload() const {
     return payload_;
@@ -338,7 +392,12 @@ C& BasicMessageBuilder<T, C>::get_concrete() {
  */
 class MessageBuilder : public BasicMessageBuilder<Buffer, MessageBuilder> {
 public:
+    using Base = BasicMessageBuilder<Buffer, MessageBuilder>;
     using BasicMessageBuilder::BasicMessageBuilder;
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    using HeaderType = Base::HeaderType;
+    using HeaderListType = Base::HeaderListType;
+#endif
 
     void construct_buffer(Buffer& lhs, const Buffer& rhs) {
         lhs = Buffer(rhs.get_data(), rhs.get_size());

include/cppkafka/producer.h

@@ -113,6 +113,7 @@ public:
      * \param builder The builder class used to compose a message
      */
     void produce(const MessageBuilder& builder);
+    void produce(MessageBuilder&& builder);
     
     /**
      * \brief Produces a message
@@ -120,6 +121,7 @@ public:
      * \param message The message to be produced
      */
     void produce(const Message& message);
+    void produce(Message&& message);
 
     /**
      * \brief Polls on this handle
@@ -157,6 +159,15 @@ public:
      */
     void flush(std::chrono::milliseconds timeout);
 private:
+#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
+    void do_produce(const MessageBuilder& builder, MessageBuilder::HeaderListType&& headers);
+    void do_produce(const Message& message, MessageBuilder::HeaderListType&& headers);
+#else
+    void do_produce(const MessageBuilder& builder);
+    void do_produce(const Message& message);
+#endif
+    
+    // Members
     PayloadPolicy message_payload_policy_;
 };