#include <followJsonRequestProvider.h>

Inheritance diagram for followJsonRequestProvider:

Collaboration diagram for followJsonRequestProvider:

Public Member Functions
	followJsonRequestProvider (decltype(requests) aRequests, decltype(delayedRequests) aDelayedRequests, decltype(parents) aParents, decltype(InputHandler) aInputHandler, decltype(OutputHandler) aOutputHandler)

	~followJsonRequestProvider () override=default

Public Member Functions inherited from followRequestProvider
	~followRequestProvider () override=default

bool	isFollowMode () const override

void	processSources (const std::vector< std::string > &sources) override

Public Member Functions inherited from requestProvider
	requestProvider (decltype(requests) aRequests, decltype(delayedRequests) aDelayedRequests, decltype(parents) aParents, decltype(InputHandler) aInputHandler, decltype(OutputHandler) aOutputHandler)
	! use only the file name part in the destination More...

virtual	~requestProvider ()=default

virtual void	prepareMappings (std::vector< std::string > &sources, std::string &destination)

virtual void	printMappings (std::ostream &stream)

Protected Member Functions
void	cleanupRenameEvents (copyRequest::clock_type::duration minAge)

void	followStream (std::istream &stream) override

virtual void	processJson (const Json::Value &json)

Protected Member Functions inherited from followRequestProvider
template<typename ... Types>
void	enqueueOrAppend (const std::string &srcPath, copyRequest::fileInWork::slotTypes type, copyRequest::clock_type::time_point timestamp, Types ... args)

virtual void	handleMove (const std::string &srcPath, const std::string &dstPath, const std::string &fromPath, const std::string &origPath, const singleMap &mapEntry, copyRequest::clock_type::time_point timestamp)

virtual void	handleOther (const std::string &srcPath, const std::string &dstPath, const singleMap &mapEntry, copyRequest::clock_type::time_point timestamp, bool requestForRemoval)

	followRequestProvider (decltype(requests) aRequests, decltype(delayedRequests) aDelayedRequests, decltype(parents) aParents, decltype(InputHandler) aInputHandler, decltype(OutputHandler) aOutputHandler)

Protected Member Functions inherited from requestProvider
virtual const singleMap &	getDstPath (const std::string &source, std::string &destination, bool baseNameOnly=false)
	get detstination papth for a given source path More...

Static Protected Member Functions
static std::string	getRequiredJsonValue (const Json::Value &json, const std::string &key)

Static Protected Member Functions inherited from requestProvider
static std::map< std::string, factoryClass * > &	getFactoryMap ()

Private Attributes
renameEventHandler	renameHandler

Static Private Attributes
static factoryTemplate< followJsonRequestProvider >	factory

static options::single< int >	jsonDelimiter

static options::single< std::string >	pathKey

static options::single< std::string >	operationKey

static options::single< std::string >	timestampKey

static options::single< std::string >	timestampFormat

static options::single< copyRequest::clock_type::duration >	timestampOffset

static options::single< std::string >	cookieKey

static options::single< std::string >	copyOpValue

static options::single< std::string >	deleteOpValue

static options::single< std::string >	moveFromOpValue

static options::single< std::string >	moveToOpValue

static options::single< std::regex >	sourceMatchRegex

Additional Inherited Members
Static Public Member Functions inherited from requestProvider
static requestProvider *	newProvider (const std::string &choice, decltype(requests) aRequests, decltype(delayedRequests) aDelayedRequests, decltype(parents) aParents, decltype(InputHandler) aInputHandler, decltype(OutputHandler) aOutputHandler)

static void	addAllowedNamesToOption (options::single< std::string > &option)

Protected Attributes inherited from requestProvider
copyRequest::simpleQueue &	requests

copyRequest::timedQueue &	delayedRequests

bool	parents

inputHandler::base *	InputHandler

outputHandler::base *	OutputHandler

Static Protected Attributes inherited from followRequestProvider
static options::single< bool >	mayDelete

static options::single< bool >	nullDelimiter

Static Protected Attributes inherited from requestProvider
static options::map< std::string, pathMapType >	pathMap

static singleMap	badMapEntry

Detailed Description

Definition at line 9 of file followJsonRequestProvider.h.

Constructor & Destructor Documentation

◆ followJsonRequestProvider()

followJsonRequestProvider::followJsonRequestProvider	(	decltype(requests)	aRequests,
		decltype(delayedRequests)	aDelayedRequests,
		decltype(parents)	aParents,
		decltype(InputHandler)	aInputHandler,
		decltype(OutputHandler)	aOutputHandler
	)

inline

Definition at line 34 of file followJsonRequestProvider.h.

                                                                      :
         followRequestProvider(aRequests, aDelayedRequests, aParents,
                               aInputHandler, aOutputHandler) {};

◆ ~followJsonRequestProvider()

followJsonRequestProvider::~followJsonRequestProvider ( )

overridedefault

Member Function Documentation

◆ cleanupRenameEvents()

void followJsonRequestProvider::cleanupRenameEvents ( copyRequest::clock_type::duration minAge )

protected

Definition at line 155 of file followJsonRequestProvider.cpp.

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

References errMsg::debug, errMsg::emit(), renameEventHandler::forgetRenameEvent(), renameEventHandler::getNextStaleSingle(), followRequestProvider::handleOther(), followRequestProvider::mayDelete, renameHalfEvent::moveFrom, and renameHandler.

Referenced by followStream(), and followKafkaRequestProvider::processSources().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ followStream()

void followJsonRequestProvider::followStream ( std::istream & stream )

overrideprotectedvirtual

Implements followRequestProvider.

Reimplemented in followKafkaRequestProvider.

Definition at line 37 of file followJsonRequestProvider.cpp.

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

References getlineWithTimeout::active(), cleanupRenameEvents(), options::base::fIsSet(), jsonDelimiter, getlineWithTimeout::nocopyLine::line(), followRequestProvider::nullDelimiter, processJson(), stopRequest::Requested(), and getlineWithTimeout::nocopyLine::timedOut().

Here is the call graph for this function:

◆ getRequiredJsonValue()

std::string followJsonRequestProvider::getRequiredJsonValue	(	const Json::Value &	json,
		const std::string &	key
	)

staticprotected

Definition at line 68 of file followJsonRequestProvider.cpp.

                               {
     auto jsonValue = json.find(key.c_str(), key.c_str() + key.size());
     if (jsonValue == nullptr) {
         return "";
     }
     return jsonValue->asString();
 }

Referenced by processJson().

Here is the caller graph for this function:

◆ processJson()

void followJsonRequestProvider::processJson ( const Json::Value & json )

protectedvirtual

Definition at line 77 of file followJsonRequestProvider.cpp.

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