followKafkaRequestProvider.cpp

Go to the documentation of this file.

#include "followKafkaRequestProvider.h"
#include <json/json.h>
#include "errMsgQueue.h"
 
defineStatic(followKafkaRequestProvider::factory, "KafkaStream");
defineStatic(followKafkaRequestProvider::kafkaInQueue, "kafkaIn");
 
void followKafkaRequestProvider::followStream(std::istream& /*stream*/) {
}
 
void followKafkaRequestProvider::processSources(const std::vector<std::string>& /*sources*/) {
    Json::Value json;
    while (true) {
        try {
            cleanupRenameEvents(std::chrono::seconds(30));
            if (stopRequest::Requested()) {
                break;
            }
            auto message = kafkaInQueue.receive(std::chrono::seconds(10));
            if (! message.empty()) {
                std::stringstream buffer(message);
                try {
                    buffer >> json;
                    processJson(json);
                } catch (const std::exception& e) {
                    errMsg::emit(errMsg::level::debug, errMsg::location(),
                                 "kafka receiver", "json decode", e.what(), " '",  message, "'");
                }
                json.clear();
            }
        } catch (const std::exception& e) {
            errMsg::emit(errMsg::level::info, errMsg::location(),
                         "kafka receiver", "receive", e.what());
            break;
        }
    }
    cleanupRenameEvents(std::chrono::seconds(-10));
}