followJsonRequestProvider.cpp

Go to the documentation of this file.

#include <OptionsChrono.h>
#include "followJsonRequestProvider.h"
#include "getlineWithTimeout.h"
#include <json/json.h>
#include <time.h>
 
 
 
defineStatic(followJsonRequestProvider::factory, "JsonStream");
 
defineStatic(followJsonRequestProvider::jsonDelimiter, '\0', "delimiter",
             "delimiter between json records, if not \\0 or \\n", 0);
defineStatic(followJsonRequestProvider::pathKey, '\0', "jsonPathKey",
             "key for path data in json input", "path");
defineStatic(followJsonRequestProvider::operationKey, '\0', "jsonOperationKey",
             "key for operation in json input", "event");
defineStatic(followJsonRequestProvider::timestampKey, '\0', "jsonTimestampKey",
             "key for timestamp in json input", "eventTime");
defineStatic(followJsonRequestProvider::timestampFormat, '\0', "jsonTimestampFormat",
             "format for timestamp in json input, like strptime",
             "%Y-%m-%d_%H:%M:%S%z");
defineStatic(followJsonRequestProvider::timestampOffset, '\0', "jsonTimestampOffset",
             "offset of timestamp", copyRequest::clock_type::duration::zero());
defineStatic(followJsonRequestProvider::cookieKey, '\0', "jsonCookieKey",
             "key for rename cookie  in json input", "cookie");
defineStatic(followJsonRequestProvider::copyOpValue, '\0', "jsonCopyOp",
             "operation value triggering a copy", "IN_CLOSE_WRITE");
defineStatic(followJsonRequestProvider::deleteOpValue, '\0', "jsonDeleteOp",
             "operation value triggering a delete", "IN_DELETE");
defineStatic(followJsonRequestProvider::moveFromOpValue, '\0', "jsonMoveFromOp",
             "operation value triggering a move from", "IN_MOVED_FROM");
defineStatic(followJsonRequestProvider::moveToOpValue, '\0', "jsonMoveToOp",
             "operation value triggering a move to", "IN_MOVED_TO");
defineStatic(followJsonRequestProvider::sourceMatchRegex, '\0', "jsonSourceMatchRegex",
             "regegx that must be matched if set to consider a path");
 
void followJsonRequestProvider::followStream(std::istream& stream) {
    char delimiter = nullDelimiter ? '\0' : '\n';
    if (jsonDelimiter.fIsSet()) {
        delimiter = jsonDelimiter;
    }
    Json::Value json;
    getlineWithTimeout lineGetter(stream, delimiter);
    while (lineGetter.active()) {
        cleanupRenameEvents(std::chrono::seconds(30));
        if (stopRequest::Requested()) {
            break;
        }
        {
            // scope to contain line and it's lock
            getlineWithTimeout::nocopyLine line(lineGetter, std::chrono::seconds(5));
            if (line.timedOut()) {
                continue;
            }
            if (line.line().empty()) {
                continue;
            }
            std::stringstream buffer(line.line());
            buffer >> json;
        }
        processJson(json);
        json.clear();
    }
    cleanupRenameEvents(std::chrono::seconds(-10));
    lineGetter.active();
}
 
std::string followJsonRequestProvider::getRequiredJsonValue(const Json::Value& json,
        const std::string& key) {
    auto jsonValue = json.find(key.c_str(), key.c_str() + key.size());
    if (jsonValue == nullptr) {
        return "";
    }
    return jsonValue->asString();
}
 
void followJsonRequestProvider::processJson(const Json::Value& json) {
    auto srcPath = getRequiredJsonValue(json, pathKey);
    if (srcPath.empty()) { // ignore json without path data
        return;
    }
    if (sourceMatchRegex.fIsSet() && ! std::regex_match(srcPath, sourceMatchRegex)) {
        return;
    }
    auto operation = getRequiredJsonValue(json, operationKey);
    if (operation.empty()) { // ignore json without operation defined
        return;
    }
 
    copyRequest::clock_type::time_point timestamp;
    auto jsonTimestamp = json.find(timestampKey.c_str(), timestampKey.c_str() + timestampKey.size());
    if (jsonTimestamp) {
        if (timestampFormat == "%s") { // use
            timestamp = copyRequest::clock_type::time_point(std::chrono::duration_cast<copyRequest::clock_type::duration>(std::chrono::duration<double>(jsonTimestamp->asDouble())));
        } else {
            struct tm tmStruct{};
            strptime(jsonTimestamp->asCString(), timestampFormat.c_str(), &tmStruct);
            timestamp = copyRequest::clock_type::time_point(std::chrono::duration_cast<copyRequest::clock_type::duration>(std::chrono::duration<double>(mktime(&tmStruct)))) + timestampOffset;
            if (timestampFormat.back() == 'z') {
                auto tzStart=jsonTimestamp->asString().find_last_of("+-");
                if (tzStart != std::string::npos) {
                    auto offset = std::stoi(jsonTimestamp->asString().substr(tzStart));
                    timestamp -= std::chrono::hours(offset/100);
                    timestamp -= std::chrono::minutes(offset%100);
                }
            }
        }
    } else {
        timestamp = copyRequest::clock_type::now();
    }
 
    std::string dstPath;
    auto mapEntry = getDstPath(srcPath, dstPath);
    if (operation == copyOpValue) {
        handleOther(srcPath, dstPath, mapEntry, timestamp, false);
    } else if (operation == deleteOpValue) {
        handleOther(srcPath, dstPath, mapEntry, timestamp, true);
    } else if (operation == moveFromOpValue || operation == moveToOpValue) {
        auto jsonCookie = getRequiredJsonValue(json, cookieKey);
        if (jsonCookie.empty()) {
            return;
        }
        renameHalfEvent::cookieType cookie = std::stoull(jsonCookie);
        renameHalfEvent::moveTypeEnum moveType;
        if (operation == moveFromOpValue) {
            moveType = renameHalfEvent::moveTypeEnum::moveFrom;
        } else {
            moveType = renameHalfEvent::moveTypeEnum::moveTo;
        }
        auto partner = renameHandler.getRenamePartner(cookie, srcPath, dstPath, moveType, mapEntry, timestamp);
        if (partner != nullptr) { // we found a partner
            if (moveType == renameHalfEvent::moveTypeEnum::moveFrom) {
                handleMove(partner->srcPath,
                           partner->dstPath,
                           dstPath,
                           srcPath,
                           mapEntry,
                           timestamp);
            } else {
                handleMove(srcPath,
                           dstPath,
                           partner->dstPath,
                           partner->srcPath,
                           mapEntry,
                           timestamp);
            }
            renameHandler.forgetRenameEvent(partner);
        }
    }
}
 
 
 
 
void followJsonRequestProvider::cleanupRenameEvents(copyRequest::clock_type::duration minAge) {
    while (auto event = renameHandler.getNextStaleSingle(minAge)) {
        if (event->moveType == renameHalfEvent::moveTypeEnum::moveFrom) {
            errMsg::emit(errMsg::level::debug, errMsg::location(),
                         event->dstPath, "found stale moveFrom, cookie", event->cookie,
                         "from", std::fixed, event->timestamp);
            if (mayDelete) {
                handleOther(event->srcPath, event->dstPath, event->mapEntry, event->timestamp, true);
            }
        } else {
            errMsg::emit(errMsg::level::debug, errMsg::location(),
                         event->dstPath, "found stale moveTo, cookie", event->cookie,
                         "from", std::fixed, event->timestamp);
            handleOther(event->srcPath, event->dstPath, event->mapEntry, event->timestamp, false);
        }
        renameHandler.forgetRenameEvent(event);
    }
}