ProducerCommon.h

#pragma once
 
#include <kafka/Project.h>
 
#include <kafka/ProducerRecord.h>
#include <kafka/RdKafkaHelper.h>
#include <kafka/Timestamp.h>
#include <kafka/Types.h>
 
#include <librdkafka/rdkafka.h>
 
#include <functional>
#include <memory>
 
 
namespace KAFKA_API { namespace clients { namespace producer {
 
class RecordMetadata
{
public:
    enum class PersistedStatus { Not, Possibly, Done };
 
    RecordMetadata() = default;
 
    RecordMetadata(const RecordMetadata& another) { *this = another; }
 
    // This is only called by the KafkaProducer::deliveryCallback (with a valid rkmsg pointer)
    RecordMetadata(const rd_kafka_message_t* rkmsg, Optional<ProducerRecord::Id> recordId)
        : _rkmsg(rkmsg), _recordId(recordId) {}
 
    RecordMetadata& operator=(const RecordMetadata& another)
    {
        if (this != &another)
        {
            const auto offsetOption = another.offset();
            _cachedInfo = std::make_unique<CachedInfo>(another.topic(),
                                                       another.partition(),
                                                       offsetOption ? *offsetOption : RD_KAFKA_OFFSET_INVALID,
                                                       another.keySize(),
                                                       another.valueSize(),
                                                       another.timestamp(),
                                                       another.persistedStatus());
            _recordId = another._recordId;
            _rkmsg    = nullptr;
        }
 
        return *this;
    }
 
    std::string                  topic()      const
    {
        return _rkmsg ? (_rkmsg->rkt ? rd_kafka_topic_name(_rkmsg->rkt) : "") : _cachedInfo->topic;
    }
 
    Partition                    partition()  const
    {
        return _rkmsg ? _rkmsg->partition : _cachedInfo->partition;
    }
 
    Optional<Offset>             offset()     const
    {
        auto offset = _rkmsg ? _rkmsg->offset : _cachedInfo->offset;
        return (offset != RD_KAFKA_OFFSET_INVALID) ? Optional<Offset>(offset) : Optional<Offset>();
    }
 
    Optional<ProducerRecord::Id> recordId()   const
    {
        return _recordId;
    }
 
    KeySize                      keySize()    const
    {
        return _rkmsg ? _rkmsg->key_len : _cachedInfo->keySize;
    }
 
    ValueSize                    valueSize()  const
    {
        return _rkmsg ? _rkmsg->len : _cachedInfo->valueSize;
    }
 
    Timestamp                    timestamp()  const
    {
        return _rkmsg ? getMsgTimestamp(_rkmsg) : _cachedInfo->timestamp;
    }
 
    PersistedStatus              persistedStatus()  const
    {
        return _rkmsg ? getMsgPersistedStatus(_rkmsg) : _cachedInfo->persistedStatus;
    }
 
    std::string                  persistedStatusString() const
    {
        return getPersistedStatusString(persistedStatus());
    }
 
    std::string                  toString() const
    {
        const auto offsetOption   = offset();
        const auto recordIdOption = recordId();
 
        return topic() + "-" + std::to_string(partition()) + "@" + (offsetOption ? std::to_string(*offsetOption) : "NA")
               + (recordIdOption ? (":id[" + std::to_string(*recordIdOption) + "],") : ",")
               + timestamp().toString() + "," + persistedStatusString();
    }
 
private:
    static Timestamp getMsgTimestamp(const rd_kafka_message_t* rkmsg)
    {
        rd_kafka_timestamp_type_t tstype{};
        const Timestamp::Value tsValue = rd_kafka_message_timestamp(rkmsg, &tstype);
        return {tsValue, tstype};
    }
 
    static PersistedStatus getMsgPersistedStatus(const rd_kafka_message_t* rkmsg)
    {
        const rd_kafka_msg_status_t status = rd_kafka_message_status(rkmsg);
        return status == RD_KAFKA_MSG_STATUS_NOT_PERSISTED ? PersistedStatus::Not : (status == RD_KAFKA_MSG_STATUS_PERSISTED ? PersistedStatus::Done : PersistedStatus::Possibly);
    }
 
    static std::string getPersistedStatusString(PersistedStatus status)
    {
        return status == PersistedStatus::Not ? "NotPersisted" :
            (status == PersistedStatus::Done ? "Persisted" : "PossiblyPersisted");
    }
 
    struct CachedInfo
    {
        CachedInfo(Topic t, Partition p, Offset o, KeySize ks, ValueSize vs, Timestamp ts, PersistedStatus pst)
            : topic(std::move(t)),
              partition(p),
              offset(o),
              keySize(ks),
              valueSize(vs),
              timestamp(ts),
              persistedStatus(pst)
        {
        }
 
        CachedInfo(const CachedInfo&) = default;
 
        std::string     topic;
        Partition       partition;
        Offset          offset;
        KeySize         keySize;
        ValueSize       valueSize;
        Timestamp       timestamp;
        PersistedStatus persistedStatus;
    };
 
    std::unique_ptr<CachedInfo>  _cachedInfo;
    const rd_kafka_message_t*    _rkmsg    = nullptr;
    Optional<ProducerRecord::Id> _recordId;
};
 
using Callback = std::function<void(const RecordMetadata& metadata, const Error& error)>;
 
} } } // end of KAFKA_API::clients::producer