inc/Processor.h - nifi-minifi-cpp - Git at Google

 /**
  * @file Processor.h
  * Processor class declaration
  *
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 #ifndef __PROCESSOR_H__
 #define __PROCESSOR_H__

 #include <uuid/uuid.h>
 #include <vector>
 #include <queue>
 #include <map>
 #include <mutex>
 #include <atomic>
 #include <algorithm>
 #include <set>

 #include "TimeUtil.h"
 #include "Property.h"
 #include "Relationship.h"
 #include "Connection.h"

 //! Forwarder declaration
 class ProcessContext;
 class ProcessSession;

 //! Minimum scheduling period in Nano Second
 #define MINIMUM_SCHEDULING_NANOS 30000

 //! Default yield period in second
 #define DEFAULT_YIELD_PERIOD_SECONDS 1

 //! Default penalization period in second
 #define DEFAULT_PENALIZATION_PERIOD_SECONDS 30

 /*!
  * Indicates the valid values for the state of a entity
  * with respect to scheduling the entity to run.
  */
 enum ScheduledState {

     /**
      * Entity cannot be scheduled to run
      */
     DISABLED,
     /**
      * Entity can be scheduled to run but currently is not
      */
     STOPPED,
     /**
      * Entity is currently scheduled to run
      */
     RUNNING
 };

 /*!
  * Scheduling Strategy
  */
 enum SchedulingStrategy {
 	//! Event driven
 	EVENT_DRIVEN,
 	//! Timer driven
 	TIMER_DRIVEN,
 	//! Cron Driven
 	CRON_DRIVEN
 };

 //! Processor Class
 class Processor
 {
 	friend class ProcessContext;
 public:
 	//! Constructor
 	/*!
 	 * Create a new processor
 	 */
 	Processor(std::string name, uuid_t uuid = NULL);
 	//! Destructor
 	virtual ~Processor();
 	//! Set Processor Name
 	void setName(std::string name) {
 		_name = name;
 	}
 	//! Get Process Name
 	std::string getName(void) {
 		return (_name);
 	}
 	//! Set UUID
 	void setUUID(uuid_t uuid) {
 		uuid_copy(_uuid, uuid);
 		char uuidStr[37];
 		uuid_unparse(_uuid, uuidStr);
 		_uuidStr = uuidStr;
 	}
 	//! Get UUID
 	bool getUUID(uuid_t uuid) {
 		if (uuid)
 		{
 			uuid_copy(uuid, _uuid);
 			return true;
 		}
 		else
 			return false;
 	}
 	//! Set the supported processor properties while the process is not running
 	bool setSupportedProperties(std::set<Property> properties);
 	//! Set the supported relationships while the process is not running
 	bool setSupportedRelationships(std::set<Relationship> relationships);
 	//! Get the supported property value by name
 	bool getProperty(std::string name, std::string &value);
 	//! Set the supported property value by name wile the process is not running
 	bool setProperty(std::string name, std::string value);
 	//! Whether the relationship is supported
 	bool isSupportedRelationship(Relationship relationship);
 	//! Set the auto terminated relationships while the process is not running
 	bool setAutoTerminatedRelationships(std::set<Relationship> relationships);
 	//! Check whether the relationship is auto terminated
 	bool isAutoTerminated(Relationship relationship);
 	//! Check whether the processor is running
 	bool isRunning();
 	//! Set Processor Scheduled State
 	void setScheduledState(ScheduledState state) {
 		_state = state;
 	}
 	//! Get Processor Scheduled State
 	ScheduledState getScheduledState(void) {
 		return _state;
 	}
 	//! Set Processor Scheduling Strategy
 	void setSchedulingStrategy(SchedulingStrategy strategy) {
 		_strategy = strategy;
 	}
 	//! Get Processor Scheduling Strategy
 	SchedulingStrategy getSchedulingStrategy(void) {
 		return _strategy;
 	}
 	//! Set Processor Loss Tolerant
 	void setlossTolerant(bool lossTolerant) {
 		_lossTolerant = lossTolerant;
 	}
 	//! Get Processor Loss Tolerant
 	bool getlossTolerant(void) {
 		return _lossTolerant;
 	}
 	//! Set Processor Scheduling Period in Nano Second
 	void setSchedulingPeriodNano(uint64_t period) {
 		uint64_t minPeriod = MINIMUM_SCHEDULING_NANOS;
 		_schedulingPeriodNano = std::max(period, minPeriod);
 	}
 	//! Get Processor Scheduling Period in Nano Second
 	uint64_t getSchedulingPeriodNano(void) {
 		return _schedulingPeriodNano;
 	}
 	//! Set Processor Run Duration in Nano Second
 	void setRunDurationNano(uint64_t period) {
 		_runDurantionNano = period;
 	}
 	//! Get Processor Run Duration in Nano Second
 	uint64_t getRunDurationNano(void) {
 		return(_runDurantionNano);
 	}
 	//! Set Processor yield period in MilliSecond
 	void setYieldPeriodMsec(uint64_t period) {
 		_yieldPeriodMsec = period;
 	}
 	//! Get Processor yield period in MilliSecond
 	uint64_t getYieldPeriodMsec(void) {
 		return(_yieldPeriodMsec);
 	}
 	//! Set Processor penalization period in MilliSecond
 	void setPenalizationPeriodMsec(uint64_t period) {
 		_penalizationPeriodMsec = period;
 	}
 	//! Get Processor penalization period in MilliSecond
 	uint64_t getPenalizationPeriodMsec(void) {
 		return(_penalizationPeriodMsec);
 	}
 	//! Set Processor Maximum Concurrent Tasks
 	void setMaxConcurrentTasks(uint8_t tasks) {
 		_maxConcurrentTasks = tasks;
 	}
 	//! Get Processor Maximum Concurrent Tasks
 	uint8_t getMaxConcurrentTasks(void) {
 		return(_maxConcurrentTasks);
 	}
 	//! Set Trigger when empty
 	void setTriggerWhenEmpty(bool value) {
 		_triggerWhenEmpty = value;
 	}
 	//! Get Trigger when empty
 	bool getTriggerWhenEmpty(void) {
 		return(_triggerWhenEmpty);
 	}
 	//! Get Active Task Counts
 	uint8_t getActiveTasks(void) {
 		return(_activeTasks);
 	}
 	//! Increment Active Task Counts
 	void incrementActiveTasks(void) {
 		_activeTasks++;
 	}
 	//! decrement Active Task Counts
 	void decrementActiveTask(void) {
 		_activeTasks--;
 	}
 	void clearActiveTask(void) {
 		_activeTasks = 0;
 	}
 	//! Yield based on the yield period
 	void yield()
 	{
 		_yieldExpiration = (getTimeMillis() + _yieldPeriodMsec);
 	}
 	//! Yield based on the input time
 	void yield(uint64_t time)
 	{
 		_yieldExpiration = (getTimeMillis() + time);
 	}
 	//! whether need be to yield
 	bool isYield()
 	{
 		if (_yieldExpiration > 0)
 			return (_yieldExpiration >= getTimeMillis());
 		else
 			return false;
 	}
 	// clear yield expiration
 	void clearYield()
 	{
 		_yieldExpiration = 0;
 	}
 	// get yield time
 	uint64_t getYieldTime()
 	{
 		uint64_t curTime = getTimeMillis();
 		if (_yieldExpiration > curTime)
 			return (_yieldExpiration - curTime);
 		else
 			return 0;;
 	}
 	//! Whether flow file queued in incoming connection
 	bool flowFilesQueued();
 	//! Whether flow file queue full in any of the outgoin connection
 	bool flowFilesOutGoingFull();
 	//! Get incoming connections
 	std::set<Connection *> getIncomingConnections() {
 		return _incomingConnections;
 	}
 	//! Has Incoming Connection
 	bool hasIncomingConnections() {
 		return (_incomingConnections.size() > 0);
 	}
 	//! Get outgoing connections based on relationship name
 	std::set<Connection *> getOutGoingConnections(std::string relationship);
 	//! Add connection
 	bool addConnection(Connection *connection);
 	//! Remove connection
 	void removeConnection(Connection *connection);
 	//! Get the UUID as string
 	std::string getUUIDStr() {
 		return _uuidStr;
 	}
 	//! Get the Next RoundRobin incoming connection
 	Connection *getNextIncomingConnection();
 	//! On Trigger
 	void onTrigger();

 public:
 	//! OnTrigger method, implemented by NiFi Processor Designer
 	virtual void onTrigger(ProcessContext *context, ProcessSession *session) = 0;
 	//! Initialize, over write by NiFi Process Designer
 	virtual void initialize(void) {
 		return;
 	}

 protected:

 	//! A global unique identifier
 	uuid_t _uuid;
 	//! Processor Name
 	std::string _name;
 	//! Supported properties
 	std::map<std::string, Property> _properties;
 	//! Supported relationships
 	std::map<std::string, Relationship> _relationships;
 	//! Autoterminated relationships
 	std::map<std::string, Relationship> _autoTerminatedRelationships;
 	//! Processor state
 	std::atomic<ScheduledState> _state;
 	//! Scheduling Strategy
 	std::atomic<SchedulingStrategy> _strategy;
 	//! lossTolerant
 	std::atomic<bool> _lossTolerant;
 	//! SchedulePeriod in Nano Seconds
 	std::atomic<uint64_t> _schedulingPeriodNano;
 	//! Run Duration in Nano Seconds
 	std::atomic<uint64_t> _runDurantionNano;
 	//! Yield Period in Milliseconds
 	std::atomic<uint64_t> _yieldPeriodMsec;
 	//! Penalization Period in MilliSecond
 	std::atomic<uint64_t> _penalizationPeriodMsec;
 	//! Maximum Concurrent Tasks
 	std::atomic<uint8_t> _maxConcurrentTasks;
 	//! Active Tasks
 	std::atomic<uint8_t> _activeTasks;
 	//! Trigger the Processor even if the incoming connection is empty
 	std::atomic<bool> _triggerWhenEmpty;
 	//! Incoming connections
 	std::set<Connection *> _incomingConnections;
 	//! Outgoing connections map based on Relationship name
 	std::map<std::string, std::set<Connection *>> _outGoingConnections;
 	//! UUID string
 	std::string _uuidStr;

 private:

 	//! Mutex for protection
 	std::mutex _mtx;
 	//! Yield Expiration
 	std::atomic<uint64_t> _yieldExpiration;
 	//! Incoming connection Iterator
 	std::set<Connection *>::iterator _incomingConnectionsIter;
 	//! Logger
 	Logger *_logger;
 	// Prevent default copy constructor and assignment operation
 	// Only support pass by reference or pointer
 	Processor(const Processor &parent);
 	Processor &operator=(const Processor &parent);

 };

 #endif
	/**
	* @file Processor.h
	* Processor class declaration
	*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	#ifndef __PROCESSOR_H__
	#define __PROCESSOR_H__

	#include <uuid/uuid.h>
	#include <vector>
	#include <queue>
	#include <map>
	#include <mutex>
	#include <atomic>
	#include <algorithm>
	#include <set>

	#include "TimeUtil.h"
	#include "Property.h"
	#include "Relationship.h"
	#include "Connection.h"

	//! Forwarder declaration
	class ProcessContext;
	class ProcessSession;

	//! Minimum scheduling period in Nano Second
	#define MINIMUM_SCHEDULING_NANOS 30000

	//! Default yield period in second
	#define DEFAULT_YIELD_PERIOD_SECONDS 1

	//! Default penalization period in second
	#define DEFAULT_PENALIZATION_PERIOD_SECONDS 30

	/*!
	* Indicates the valid values for the state of a entity
	* with respect to scheduling the entity to run.
	*/
	enum ScheduledState {

	/**
	* Entity cannot be scheduled to run
	*/
	DISABLED,
	/**
	* Entity can be scheduled to run but currently is not
	*/
	STOPPED,
	/**
	* Entity is currently scheduled to run
	*/
	RUNNING
	};

	/*!
	* Scheduling Strategy
	*/
	enum SchedulingStrategy {
	//! Event driven
	EVENT_DRIVEN,
	//! Timer driven
	TIMER_DRIVEN,
	//! Cron Driven
	CRON_DRIVEN
	};

	//! Processor Class
	class Processor
	{
	friend class ProcessContext;
	public:
	//! Constructor
	/*!
	* Create a new processor
	*/
	Processor(std::string name, uuid_t uuid = NULL);
	//! Destructor
	virtual ~Processor();
	//! Set Processor Name
	void setName(std::string name) {
	_name = name;
	}
	//! Get Process Name
	std::string getName(void) {
	return (_name);
	}
	//! Set UUID
	void setUUID(uuid_t uuid) {
	uuid_copy(_uuid, uuid);
	char uuidStr[37];
	uuid_unparse(_uuid, uuidStr);
	_uuidStr = uuidStr;
	}
	//! Get UUID
	bool getUUID(uuid_t uuid) {
	if (uuid)
	{
	uuid_copy(uuid, _uuid);
	return true;
	}
	else
	return false;
	}
	//! Set the supported processor properties while the process is not running
	bool setSupportedProperties(std::set<Property> properties);
	//! Set the supported relationships while the process is not running
	bool setSupportedRelationships(std::set<Relationship> relationships);
	//! Get the supported property value by name
	bool getProperty(std::string name, std::string &value);
	//! Set the supported property value by name wile the process is not running
	bool setProperty(std::string name, std::string value);
	//! Whether the relationship is supported
	bool isSupportedRelationship(Relationship relationship);
	//! Set the auto terminated relationships while the process is not running
	bool setAutoTerminatedRelationships(std::set<Relationship> relationships);
	//! Check whether the relationship is auto terminated
	bool isAutoTerminated(Relationship relationship);
	//! Check whether the processor is running
	bool isRunning();
	//! Set Processor Scheduled State
	void setScheduledState(ScheduledState state) {
	_state = state;
	}
	//! Get Processor Scheduled State
	ScheduledState getScheduledState(void) {
	return _state;
	}
	//! Set Processor Scheduling Strategy
	void setSchedulingStrategy(SchedulingStrategy strategy) {
	_strategy = strategy;
	}
	//! Get Processor Scheduling Strategy
	SchedulingStrategy getSchedulingStrategy(void) {
	return _strategy;
	}
	//! Set Processor Loss Tolerant
	void setlossTolerant(bool lossTolerant) {
	_lossTolerant = lossTolerant;
	}
	//! Get Processor Loss Tolerant
	bool getlossTolerant(void) {
	return _lossTolerant;
	}
	//! Set Processor Scheduling Period in Nano Second
	void setSchedulingPeriodNano(uint64_t period) {
	uint64_t minPeriod = MINIMUM_SCHEDULING_NANOS;
	_schedulingPeriodNano = std::max(period, minPeriod);
	}
	//! Get Processor Scheduling Period in Nano Second
	uint64_t getSchedulingPeriodNano(void) {
	return _schedulingPeriodNano;
	}
	//! Set Processor Run Duration in Nano Second
	void setRunDurationNano(uint64_t period) {
	_runDurantionNano = period;
	}
	//! Get Processor Run Duration in Nano Second
	uint64_t getRunDurationNano(void) {
	return(_runDurantionNano);
	}
	//! Set Processor yield period in MilliSecond
	void setYieldPeriodMsec(uint64_t period) {
	_yieldPeriodMsec = period;
	}
	//! Get Processor yield period in MilliSecond
	uint64_t getYieldPeriodMsec(void) {
	return(_yieldPeriodMsec);
	}
	//! Set Processor penalization period in MilliSecond
	void setPenalizationPeriodMsec(uint64_t period) {
	_penalizationPeriodMsec = period;
	}
	//! Get Processor penalization period in MilliSecond
	uint64_t getPenalizationPeriodMsec(void) {
	return(_penalizationPeriodMsec);
	}
	//! Set Processor Maximum Concurrent Tasks
	void setMaxConcurrentTasks(uint8_t tasks) {
	_maxConcurrentTasks = tasks;
	}
	//! Get Processor Maximum Concurrent Tasks
	uint8_t getMaxConcurrentTasks(void) {
	return(_maxConcurrentTasks);
	}
	//! Set Trigger when empty
	void setTriggerWhenEmpty(bool value) {
	_triggerWhenEmpty = value;
	}
	//! Get Trigger when empty
	bool getTriggerWhenEmpty(void) {
	return(_triggerWhenEmpty);
	}
	//! Get Active Task Counts
	uint8_t getActiveTasks(void) {
	return(_activeTasks);
	}
	//! Increment Active Task Counts
	void incrementActiveTasks(void) {
	_activeTasks++;
	}
	//! decrement Active Task Counts
	void decrementActiveTask(void) {
	_activeTasks--;
	}
	void clearActiveTask(void) {
	_activeTasks = 0;
	}
	//! Yield based on the yield period
	void yield()
	{
	_yieldExpiration = (getTimeMillis() + _yieldPeriodMsec);
	}
	//! Yield based on the input time
	void yield(uint64_t time)
	{
	_yieldExpiration = (getTimeMillis() + time);
	}
	//! whether need be to yield
	bool isYield()
	{
	if (_yieldExpiration > 0)
	return (_yieldExpiration >= getTimeMillis());
	else
	return false;
	}
	// clear yield expiration
	void clearYield()
	{
	_yieldExpiration = 0;
	}
	// get yield time
	uint64_t getYieldTime()
	{
	uint64_t curTime = getTimeMillis();
	if (_yieldExpiration > curTime)
	return (_yieldExpiration - curTime);
	else
	return 0;;
	}
	//! Whether flow file queued in incoming connection
	bool flowFilesQueued();
	//! Whether flow file queue full in any of the outgoin connection
	bool flowFilesOutGoingFull();
	//! Get incoming connections
	std::set<Connection *> getIncomingConnections() {
	return _incomingConnections;
	}
	//! Has Incoming Connection
	bool hasIncomingConnections() {
	return (_incomingConnections.size() > 0);
	}
	//! Get outgoing connections based on relationship name
	std::set<Connection *> getOutGoingConnections(std::string relationship);
	//! Add connection
	bool addConnection(Connection *connection);
	//! Remove connection
	void removeConnection(Connection *connection);
	//! Get the UUID as string
	std::string getUUIDStr() {
	return _uuidStr;
	}
	//! Get the Next RoundRobin incoming connection
	Connection *getNextIncomingConnection();
	//! On Trigger
	void onTrigger();

	public:
	//! OnTrigger method, implemented by NiFi Processor Designer
	virtual void onTrigger(ProcessContext context, ProcessSession session) = 0;
	//! Initialize, over write by NiFi Process Designer
	virtual void initialize(void) {
	return;
	}

	protected:

	//! A global unique identifier
	uuid_t _uuid;
	//! Processor Name
	std::string _name;
	//! Supported properties
	std::map<std::string, Property> _properties;
	//! Supported relationships
	std::map<std::string, Relationship> _relationships;
	//! Autoterminated relationships
	std::map<std::string, Relationship> _autoTerminatedRelationships;
	//! Processor state
	std::atomic<ScheduledState> _state;
	//! Scheduling Strategy
	std::atomic<SchedulingStrategy> _strategy;
	//! lossTolerant
	std::atomic<bool> _lossTolerant;
	//! SchedulePeriod in Nano Seconds
	std::atomic<uint64_t> _schedulingPeriodNano;
	//! Run Duration in Nano Seconds
	std::atomic<uint64_t> _runDurantionNano;
	//! Yield Period in Milliseconds
	std::atomic<uint64_t> _yieldPeriodMsec;
	//! Penalization Period in MilliSecond
	std::atomic<uint64_t> _penalizationPeriodMsec;
	//! Maximum Concurrent Tasks
	std::atomic<uint8_t> _maxConcurrentTasks;
	//! Active Tasks
	std::atomic<uint8_t> _activeTasks;
	//! Trigger the Processor even if the incoming connection is empty
	std::atomic<bool> _triggerWhenEmpty;
	//! Incoming connections
	std::set<Connection *> _incomingConnections;
	//! Outgoing connections map based on Relationship name
	std::map<std::string, std::set<Connection *>> _outGoingConnections;
	//! UUID string
	std::string _uuidStr;

	private:

	//! Mutex for protection
	std::mutex _mtx;
	//! Yield Expiration
	std::atomic<uint64_t> _yieldExpiration;
	//! Incoming connection Iterator
	std::set<Connection *>::iterator _incomingConnectionsIter;
	//! Logger
	Logger *_logger;
	// Prevent default copy constructor and assignment operation
	// Only support pass by reference or pointer
	Processor(const Processor &parent);
	Processor &operator=(const Processor &parent);

	};

	#endif