messageQueue::kafka Class Reference

#include <kafkaQueue.h>

Inheritance diagram for messageQueue::kafka:

Collaboration diagram for messageQueue::kafka:

Public Member Functions
	kafka (const std::string &aName)
void	init (bool isConsumer=false)
	~kafka () override
void	send (const std::string &aMessage, const std::string &aTopic="") override
std::string	receive (std::chrono::system_clock::duration timeout) override
bool	isSenderConfigured () const override
bool	isReceiverConfigured () const override
Public Member Functions inherited from messageQueue::queue
	queue (const std::string &aName)
virtual	~queue ()=default

Private Attributes
std::map< std::string, std::unique_ptr< RdKafka::Topic > >	topics
RdKafka::Producer *	producer
RdKafka::Consumer *	consumer
RdKafka::Conf *	tconf
options::single< std::string >	topicName
options::map< std::string >	producerCfgStrings
options::map< std::string >	topicCfgStrings

Detailed Description

Definition at line 14 of file kafkaQueue.h.

Constructor & Destructor Documentation

kafka()

messageQueue::kafka::kafka

(

const std::string &

aName

)

Definition at line 6 of file kafkaQueue.cpp.

                                      :
        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") {
    }

~kafka()

messageQueue::kafka::~kafka ( )

override

Definition at line 49 of file kafkaQueue.cpp.

                  {
        if (producer) {
            producer->flush(1000);
        }
        delete producer;
        if (consumer && !topics.empty()) {
            consumer->stop(topics.begin()->second.get(), 0);
        }
        delete consumer;
    }

References consumer, producer, and topics.

Member Function Documentation

init()

void messageQueue::kafka::init

(

bool

isConsumer = false

)

Definition at line 14 of file kafkaQueue.cpp.

                                    {
        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));
                }
            }
        }
    }

References consumer, options::base::fIsSet(), producer, producerCfgStrings, tconf, topicCfgStrings, topicName, and topics.

Referenced by followKafkaRequestProvider::followKafkaRequestProvider().

Here is the call graph for this function:

Here is the caller graph for this function:

isReceiverConfigured()

bool messageQueue::kafka::isReceiverConfigured ( ) const

overridevirtual

Implements messageQueue::queue.

Definition at line 106 of file kafkaQueue.cpp.

                                           {
        return consumer != nullptr;
    }

References consumer.

isSenderConfigured()

bool messageQueue::kafka::isSenderConfigured ( ) const

overridevirtual

Implements messageQueue::queue.

Definition at line 103 of file kafkaQueue.cpp.

                                         {
        return producer != nullptr;
    }

References producer.

receive()

std::string messageQueue::kafka::receive ( std::chrono::system_clock::duration  timeout )

overridevirtual

Implements messageQueue::queue.

Definition at line 84 of file kafkaQueue.cpp.

                                                                  {
        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 "";
    }

References consumer, and topics.

Referenced by followKafkaRequestProvider::processSources().

Here is the caller graph for this function:

send()

void messageQueue::kafka::send ( const std::string &  aMessage, const std::string &  aTopic = ""  )

overridevirtual

Implements messageQueue::queue.

Definition at line 59 of file kafkaQueue.cpp.

                                              {
        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);
    }

References errMsg::debug, errMsg::emit(), producer, tconf, topicName, and topics.

Here is the call graph for this function:

Member Data Documentation

consumer

RdKafka::Consumer* messageQueue::kafka::consumer

private

Definition at line 18 of file kafkaQueue.h.

Referenced by init(), isReceiverConfigured(), receive(), and ~kafka().

producer

RdKafka::Producer* messageQueue::kafka::producer

private

Definition at line 17 of file kafkaQueue.h.

Referenced by init(), isSenderConfigured(), send(), and ~kafka().

producerCfgStrings

options::map<std::string> messageQueue::kafka::producerCfgStrings

private

Definition at line 21 of file kafkaQueue.h.

Referenced by init().

tconf

RdKafka::Conf* messageQueue::kafka::tconf

private

Definition at line 19 of file kafkaQueue.h.

Referenced by init(), and send().

topicCfgStrings

options::map<std::string> messageQueue::kafka::topicCfgStrings

private

Definition at line 22 of file kafkaQueue.h.

Referenced by init().

topicName

options::single<std::string> messageQueue::kafka::topicName

private

Definition at line 20 of file kafkaQueue.h.

Referenced by init(), and send().

topics

std::map<std::string, std::unique_ptr<RdKafka::Topic> > messageQueue::kafka::topics

private

Definition at line 16 of file kafkaQueue.h.

Referenced by init(), receive(), send(), and ~kafka().

---

The documentation for this class was generated from the following files:

• src/kafkaQueue.h (v0.19-24-g0617ca1 with changes)
• src/kafkaQueue.cpp (v0.19-24-g0617ca1 with changes)