kafkaQueue.cpp

Go to the documentation of this file.

#include "kafkaQueue.h"
#include <librdkafka/rdkafkacpp.h>
#include <errMsgQueue.h>
 
namespace messageQueue {
    kafka::kafka(const std::string& aName):
        queue(aName),
        producer(nullptr),
        consumer(nullptr),
        topicName('\0', aName + "-topic", aName + " queue topic name"),
        producerCfgStrings('\0', aName + "-pcfg", aName + " producer/consumer config strings"),
        topicCfgStrings('\0', aName + "-tcfg", aName + " topic config strings") {
    }
    void kafka::init(bool isConsumer) {
        if (! producerCfgStrings.empty()) {
            std::string errstr; // evil expensive reporting...
            RdKafka::Conf *pconf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
            for (const auto& item : producerCfgStrings) {
                if (pconf->set(item.first, item.second, errstr) != RdKafka::Conf::CONF_OK) {
                    throw std::runtime_error(errstr);
                }
            }
            tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
            for (const auto& item : topicCfgStrings) {
                if (tconf->set(item.first, item.second, errstr) != RdKafka::Conf::CONF_OK) {
                    throw std::runtime_error(errstr);
                }
            }
            if (isConsumer) {
                consumer = RdKafka::Consumer::create(pconf, errstr);
                decltype(topics)::mapped_type topic(RdKafka::Topic::create(consumer, topicName, tconf, errstr));
                if (topic == nullptr) {
                    throw std::runtime_error(errstr);
                }
                topics.emplace(topicName, std::move(topic));
                consumer->start(topics.begin()->second.get(), 0, RdKafka::Topic::OFFSET_END);
            } else {
                producer = RdKafka::Producer::create(pconf, errstr);
                if (topicName.fIsSet()) {
                    decltype(topics)::mapped_type topic(RdKafka::Topic::create(producer, topicName, tconf, errstr));
                    if (topic == nullptr) {
                        throw std::runtime_error(errstr);
                    }
                    topics.emplace(topicName, std::move(topic));
                }
            }
        }
    }
    kafka::~kafka() {
        if (producer) {
            producer->flush(1000);
        }
        delete producer;
        if (consumer && !topics.empty()) {
            consumer->stop(topics.begin()->second.get(), 0);
        }
        delete consumer;
    }
    void kafka::send(const std::string& aMessage,
                     const std::string& aTopic) {
        RdKafka::Topic* topic;
        if (aTopic.empty()) {
            topic = topics[topicName].get();
        } else {
            auto it = topics.find(aTopic);
            if (it == topics.end()) {
                std::string errstr;
                decltype(topics)::mapped_type newTopic(RdKafka::Topic::create(producer, aTopic, tconf, errstr));
                if (newTopic == nullptr) {
                    throw std::runtime_error(errstr);
                }
                auto result = topics.emplace(aTopic, std::move(newTopic));
                errMsg::emit(errMsg::level::debug, errMsg::location(),
                             aTopic, "kafkaSender", "new dynamic topic created");
                it = result.first;
            }
            topic = it->second.get();
        }
        producer->produce(topic, RdKafka::Topic::PARTITION_UA,
                          RdKafka::Producer::RK_MSG_COPY,
                          const_cast<void*>(static_cast<const void *>(aMessage.data())), aMessage.size(),
                          nullptr, nullptr);
    }
    std::string kafka::receive(std::chrono::system_clock::duration timeout) {
        auto message = consumer->consume(topics.begin()->second.get(), 0,
                                         std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count());
        if (message->err() == RdKafka::ERR_NO_ERROR) {
            auto retval =  std::string(reinterpret_cast<char *>(message->payload()),
                                       message->len());
            delete message;
            return retval;
        } else if (message->err() != RdKafka::ERR_REQUEST_TIMED_OUT &&
                   message->err() != RdKafka::ERR__PARTITION_EOF &&
                   message->err() != RdKafka::ERR__TIMED_OUT) {
            throw std::runtime_error(RdKafka::err2str(message->err())
                                     + std::to_string(message->err()));
        }
        delete message;
        return "";
    }
 
 
    bool kafka::isSenderConfigured() const {
        return producer != nullptr;
    }
    bool kafka::isReceiverConfigured() const {
        return consumer != nullptr;
    }
} // end namespace messageQueue