BrokerMetadata.h

#pragma once
 
#include <kafka/Project.h>
 
#include <kafka/KafkaException.h>
#include <kafka/Types.h>
 
#include <librdkafka/rdkafka.h>
 
#include <map>
#include <vector>
 
 
namespace KAFKA_API {
 
struct BrokerMetadata {
    struct Node
    {
      public:
        using Id   = int;
        using Host = std::string;
        using Port = int;
 
        Node(Id i, Host h, Port p): id(i), host(std::move(h)), port(p) {}
 
        Node::Id  id;
 
        Node::Host host;
 
        Node::Port port;
 
        std::string toString() const { return host + ":" + std::to_string(port) + "/" + std::to_string(id); }
    };
 
    struct PartitionInfo
    {
        explicit PartitionInfo(Node::Id leaderId): leader(leaderId) {}
 
        void addReplica(Node::Id id)       { replicas.emplace_back(id); }
        void addInSyncReplica(Node::Id id) { inSyncReplicas.emplace_back(id); }
 
        Node::Id              leader;
 
        std::vector<Node::Id> replicas;
 
        std::vector<Node::Id> inSyncReplicas;
 
    };
 
    std::string getNodeDescription(Node::Id id) const;
 
    std::string toString(const PartitionInfo& partitionInfo) const;
 
    explicit BrokerMetadata(Topic topic): _topic(std::move(topic)) {}
 
    const std::string& topic() const { return _topic; }
 
    std::vector<std::shared_ptr<Node>> nodes() const;
 
    const std::map<Partition, PartitionInfo>& partitions() const { return _partitions; }
 
    std::string toString()   const;
 
    void setOrigNodeName(const std::string& origNodeName)                          { _origNodeName = origNodeName; }
    void addNode(Node::Id nodeId, const Node::Host& host, Node::Port port)         { _nodes[nodeId] = std::make_shared<Node>(nodeId, host, port); }
    void addPartitionInfo(Partition partition, const PartitionInfo& partitionInfo) { _partitions.emplace(partition, partitionInfo); }
 
private:
    Topic                                     _topic;
    std::string                               _origNodeName;
    std::map<Node::Id, std::shared_ptr<Node>> _nodes;
    std::map<Partition, PartitionInfo>        _partitions;
};
 
inline std::vector<std::shared_ptr<BrokerMetadata::Node>>
BrokerMetadata::nodes() const
{
    std::vector<std::shared_ptr<BrokerMetadata::Node>> ret;
    ret.reserve(_nodes.size());
    for (const auto& nodeInfo: _nodes)
    {
        ret.emplace_back(nodeInfo.second);
    }
    return ret;
}
 
inline std::string
BrokerMetadata::getNodeDescription(Node::Id id) const
{
    const auto& found = _nodes.find(id);
    if (found == _nodes.cend()) return "-:-/" + std::to_string(id);
 
    auto node = found->second;
    return node->host + ":" + std::to_string(node->port) + "/" + std::to_string(id);
}
 
inline std::string
BrokerMetadata::toString(const PartitionInfo& partitionInfo) const
{
    std::ostringstream oss;
 
    auto streamNodes = [this](std::ostringstream& ss, const std::vector<Node::Id>& nodeIds) -> std::ostringstream& {
        bool isTheFirst = true;
        for (const auto id: nodeIds)
        {
            ss << (isTheFirst ? (isTheFirst = false, "") : ", ") << getNodeDescription(id);
        }
        return ss;
    };
 
    oss << "leader[" << getNodeDescription(partitionInfo.leader) << "], replicas[";
    streamNodes(oss, partitionInfo.replicas) << "], inSyncReplicas[";
    streamNodes(oss, partitionInfo.inSyncReplicas) << "]";
 
    return oss.str();
}
 
inline std::string
BrokerMetadata::toString() const
{
    std::ostringstream oss;
 
    oss << "originatingNode[" << _origNodeName << "], topic[" << _topic <<  "], partitions{";
    bool isTheFirst = true;
    for (const auto& partitionInfoPair: _partitions)
    {
        const Partition       partition     = partitionInfoPair.first;
        const PartitionInfo&  partitionInfo = partitionInfoPair.second;
        oss << (isTheFirst ? (isTheFirst = false, "") : "; ") << partition << ": " << toString(partitionInfo);
    }
    oss << "}";
 
    return oss.str();
}
 
} // end of KAFKA_API