#include <followKafkaRequestProvider.h>

Inheritance diagram for followKafkaRequestProvider:

Collaboration diagram for followKafkaRequestProvider:

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

	~followKafkaRequestProvider () override=default

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

Public Member Functions inherited from followJsonRequestProvider
	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)

Private Member Functions
void	followStream (std::istream &stream) override

Static Private Attributes
static factoryTemplate< followKafkaRequestProvider >	factory

static messageQueue::kafka	kafkaInQueue

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 Member Functions inherited from followJsonRequestProvider
void	cleanupRenameEvents (copyRequest::clock_type::duration minAge)

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 inherited from followJsonRequestProvider
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 ()

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 5 of file followKafkaRequestProvider.h.

Constructor & Destructor Documentation

◆ followKafkaRequestProvider()

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

inline

Definition at line 11 of file followKafkaRequestProvider.h.

                                                                       :
         followJsonRequestProvider(aRequests, aDelayedRequests, aParents,
                                   aInputHandler, aOutputHandler) {
         kafkaInQueue.init(true);
     };

References messageQueue::kafka::init(), and kafkaInQueue.

Here is the call graph for this function:

◆ ~followKafkaRequestProvider()

followKafkaRequestProvider::~followKafkaRequestProvider ( )

overridedefault

Member Function Documentation

◆ followStream()

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

overrideprivatevirtual

Reimplemented from followJsonRequestProvider.

Definition at line 8 of file followKafkaRequestProvider.cpp.

8 {

9 }

◆ processSources()

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

overridevirtual

Implements requestProvider.

Definition at line 11 of file followKafkaRequestProvider.cpp.

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