scottk/wlan-cloud-lib-cppkafka
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-cloud-lib-cppkafka.git synced 2025-11-01 02:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added constructor from another HeaderList type

Browse Source
This commit is contained in:
accelerated
2018-12-05 20:04:10 -05:00
parent fe0c7e7dd5
commit 1c80af9b68
3 changed files with 37 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
include/cppkafka/buffer.h
View File

@@ -89,7 +89,7 @@ public:
* \param last An iterator to the end of data (not included)
*/
template <typename Iter>
Buffer(const Iter first, Iter last)
Buffer(const Iter first, const Iter last)
: Buffer(&*first, std::distance(first, last)) {
}

33
include/cppkafka/header_list.h
View File

@@ -51,6 +51,9 @@ namespace cppkafka {
template <typename HeaderType>
class HeaderList {
public:
template <typename OtherHeaderType>
friend class HeaderList;
using BufferType = typename HeaderType::ValueType;
using Iterator = HeaderIterator<HeaderType>;
/**
@@ -75,6 +78,16 @@ public:
*/
explicit HeaderList(rd_kafka_headers_t* handle);
/**
* \brief Create a header list from another header list type
* \param other The other list
*/
template <typename OtherHeaderType>
HeaderList(const HeaderList<OtherHeaderType>& other);
template <typename OtherHeaderType>
HeaderList(HeaderList<OtherHeaderType>&& other);
/**
* \brief Add a header to the list. This translates to rd_kafka_header_add().
* \param header The header.
@@ -219,6 +232,20 @@ HeaderList<HeaderType>::HeaderList(rd_kafka_headers_t* handle, NonOwningTag)
assert(handle);
}
template <typename HeaderType>
template <typename OtherHeaderType>
HeaderList<HeaderType>::HeaderList(const HeaderList<OtherHeaderType>& other)
: handle_(other.handle_) {
}
template <typename HeaderType>
template <typename OtherHeaderType>
HeaderList<HeaderType>::HeaderList(HeaderList<OtherHeaderType>&& other)
: handle_(std::move(other.handle_)) {
}
// Methods
template <typename HeaderType>
Error HeaderList<HeaderType>::add(const HeaderType& header) {
@@ -279,13 +306,15 @@ bool HeaderList<HeaderType>::empty() const {
template <typename HeaderType>
typename HeaderList<HeaderType>::Iterator
HeaderList<HeaderType>::begin() const {
return empty() ? end() : Iterator(make_non_owning(handle_.get()), 0);
return empty() ? Iterator(HeaderList<HeaderType>(), 0) :
Iterator(make_non_owning(handle_.get()), 0);
}
template <typename HeaderType>
typename HeaderList<HeaderType>::Iterator
HeaderList<HeaderType>::end() const {
return Iterator(empty() ? HeaderList<HeaderType>() : make_non_owning(handle_.get()), size());
return empty() ? Iterator(HeaderList<HeaderType>(), size()) :
Iterator(make_non_owning(handle_.get()), size());
}
template <typename HeaderType>

9
include/cppkafka/message_builder.h
View File

@@ -244,6 +244,9 @@ BasicMessageBuilder<T, C>::BasicMessageBuilder(const Message& message)
: topic_(message.get_topic()),
key_(Buffer(message.get_key().get_data(), message.get_key().get_size())),
#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
//Here we must copy explicitly the Message headers since they are non-owning and this class
//assumes full ownership. Otherwise we will be holding an invalid handle when Message goes
//out of scope and rdkafka frees its resource.
header_list_(message.get_header_list() ?
HeaderListType(rd_kafka_headers_copy(message.get_header_list().get_handle())) : HeaderListType()), //copy headers
#endif
@@ -261,8 +264,7 @@ BasicMessageBuilder<T, C>::BasicMessageBuilder(const BasicMessageBuilder<U, V>&
: topic_(rhs.topic()),
partition_(rhs.partition()),
#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
header_list_(rhs.header_list() ?
HeaderListType(rd_kafka_headers_copy(rhs.header_list().get_handle())) : HeaderListType()), //copy headers
header_list_(rhs.header_list()), //copy headers
#endif
timestamp_(rhs.timestamp()),
user_data_(rhs.user_data()),
@@ -277,8 +279,7 @@ BasicMessageBuilder<T, C>::BasicMessageBuilder(BasicMessageBuilder<U, V>&& rhs)
: topic_(rhs.topic()),
partition_(rhs.partition()),
#if (RD_KAFKA_VERSION >= RD_KAFKA_HEADERS_SUPPORT_VERSION)
header_list_(rhs.header_list() ?
HeaderListType(rhs.header_list().release_handle()) : HeaderListType()), //assume header ownership
header_list_(std::move(header_list())), //assume header ownership
#endif
timestamp_(rhs.timestamp()),
user_data_(rhs.user_data()),