ProducerRecord.h

#pragma once
 
#include <kafka/Project.h>
 
#include <kafka/Header.h>
#include <kafka/Types.h>
 
#include <librdkafka/rdkafka.h>
 
 
namespace KAFKA_API { namespace clients { namespace producer {
 
class ProducerRecord
{
public:
    using Id  = std::uint64_t;
 
    ProducerRecord(Topic topic, Partition partition, const Key& key, const Value& value)
       : _topic(std::move(topic)), _partition(partition), _key(key), _value(value) {}
 
    ProducerRecord(const Topic& topic, Partition partition, const Key& key, const Value& value, Id id)
        : ProducerRecord(topic, partition, key, value) { _id = id; }
 
    ProducerRecord(const Topic& topic, const Key& key, const Value& value)
        : ProducerRecord(topic, RD_KAFKA_PARTITION_UA, key, value) {}
 
    ProducerRecord(const Topic& topic, const Key& key, const Value& value, Id id)
        : ProducerRecord(topic, key, value) { _id = id; }
 
    const Topic& topic()  const { return _topic; }
 
    Partition partition() const { return _partition; }
 
    Key       key()       const { return _key; }
 
    Value     value()     const { return _value; }
 
    Optional<Id>   id()      const { return _id; }
 
    const Headers& headers() const { return _headers; }
 
    Headers&       headers()       { return _headers; }
 
    void setPartition(Partition partition) { _partition = partition; }
 
    void setKey(const Key& key)            { _key = key; }
 
    void setValue(const Value& value)      { _value = value; }
 
    void setId(Id id)                      { _id = id; }
 
    std::string toString() const
    {
        return _topic + "-" + (_partition == RD_KAFKA_PARTITION_UA ? "NA" : std::to_string(_partition)) + std::string(":")
            + (_id ? (std::to_string(*_id) + std::string(", ")) : " ")
            + (_headers.empty() ? "" : ("headers[" + KAFKA_API::toString(_headers) + "], "))
            + _key.toString() + std::string("/") + _value.toString();
    }
 
private:
    Topic        _topic;
    Partition    _partition;
    Key          _key;
    Value        _value;
    Headers      _headers;
    Optional<Id> _id;
};
 
} } } // end of KAFKA_API::clients::producer