ConsumerRecord.h

#pragma once
 
#include <kafka/Project.h>
 
#include <kafka/Error.h>
#include <kafka/Header.h>
#include <kafka/Timestamp.h>
#include <kafka/Types.h>
 
#include <librdkafka/rdkafka.h>
 
#include <sstream>
 
 
namespace KAFKA_API { namespace clients { namespace consumer {
 
class ConsumerRecord
{
public:
    // ConsumerRecord will take the ownership of msg (rd_kafka_message_t*)
    explicit ConsumerRecord(rd_kafka_message_t* msg): _rk_msg(msg, rd_kafka_message_destroy) {}
 
    Topic       topic()         const { return _rk_msg->rkt ? rd_kafka_topic_name(_rk_msg->rkt): ""; }
 
    Partition   partition()     const { return _rk_msg->partition; }
 
    Offset      offset()        const { return _rk_msg->offset; }
 
    Key         key()           const { return Key(_rk_msg->key, _rk_msg->key_len); }
 
    Value       value()         const { return Value(_rk_msg->payload, _rk_msg->len); }
 
    Timestamp   timestamp() const
    {
        rd_kafka_timestamp_type_t tstype{};
        const Timestamp::Value tsValue = rd_kafka_message_timestamp(_rk_msg.get(), &tstype);
        return {tsValue, tstype};
    }
 
    Headers headers() const;
 
    Header::Value lastHeaderValue(const Header::Key& key);
 
    Error error() const { return Error{_rk_msg->err}; }
 
    std::string toString() const;
 
private:
    using rd_kafka_message_shared_ptr = std::shared_ptr<rd_kafka_message_t>;
    rd_kafka_message_shared_ptr _rk_msg;
};
 
inline Headers
ConsumerRecord::headers() const
{
    Headers headers;
 
    rd_kafka_headers_t* hdrs = nullptr;
    if (rd_kafka_message_headers(_rk_msg.get(), &hdrs) != RD_KAFKA_RESP_ERR_NO_ERROR)
    {
        return headers;
    }
 
    headers.reserve(rd_kafka_header_cnt(hdrs));
 
    const char* name      = nullptr;
    const void* valuePtr  = nullptr;
    std::size_t valueSize = 0;
    for (std::size_t i = 0; !rd_kafka_header_get_all(hdrs, i, &name, &valuePtr, &valueSize); i++)
    {
        headers.emplace_back(name, Header::Value(valuePtr, valueSize));
    }
 
    return headers;
}
 
inline Header::Value
ConsumerRecord::lastHeaderValue(const Header::Key& key)
{
    rd_kafka_headers_t* hdrs = nullptr;
    if (rd_kafka_message_headers(_rk_msg.get(), &hdrs) != RD_KAFKA_RESP_ERR_NO_ERROR)
    {
        return Header::Value();
    }
 
    const void* valuePtr  = nullptr;
    std::size_t valueSize = 0;
    return (rd_kafka_header_get_last(hdrs, key.c_str(), &valuePtr, &valueSize) == RD_KAFKA_RESP_ERR_NO_ERROR) ?
           Header::Value(valuePtr, valueSize) : Header::Value();
}
 
inline std::string
ConsumerRecord::toString() const
{
    std::ostringstream oss;
    if (!error())
    {
        oss << topic() << "-" << partition() << ":" << offset() << ", " << timestamp().toString() << ", "
            << (key().size() ? (key().toString() + "/") : "") << value().toString();
    }
    else if (error().value() == RD_KAFKA_RESP_ERR__PARTITION_EOF)
    {
        oss << "EOF[" << topic() << "-" << partition() << ":" << offset() << "]";
    }
    else
    {
        oss << "ERROR[" << error().message() << ", " << topic() << "-" << partition() << ":" << offset() << "]";
    }
    return oss.str();
}
 
} } } // end of KAFKA_API::clients::consumer