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

 /**
  * @file FlowControlProtocol.h
  * FlowControlProtocol 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 __FLOW_CONTROL_PROTOCOL_H__
 #define __FLOW_CONTROL_PROTOCOL_H__

 #include <stdio.h>
 #include <unistd.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <fcntl.h>
 #include <netdb.h>
 #include <string>
 #include <errno.h>
 #include <chrono>
 #include <thread>
 #include "Logger.h"
 #include "Configure.h"
 #include "Property.h"

 //! Forwarder declaration
 class FlowController;

 #define DEFAULT_NIFI_SERVER_PORT 9000
 #define DEFAULT_REPORT_INTERVAL 1000 // 1 sec
 #define MAX_READ_TIMEOUT 30000 // 30 seconds

 //! FlowControl Protocol Msg Type
 typedef enum {
 	REGISTER_REQ, // Device Register Request from device to server which contain device serial number, current running flow xml version
 	REGISTER_RESP, // Device Register Respond from server to device, may contain new flow.xml from server ask device to apply and also device report interval
 	REPORT_REQ, // Period Device Report from device to server which contain device serial number, current running flow xml name/version and other period report info
 	REPORT_RESP, // Report Respond from server to device, may ask device to update flow xml or processor property
 	MAX_FLOW_CONTROL_MSG_TYPE
 } FlowControlMsgType;

 //! FlowControl Protocol Msg Type String
 static const char *FlowControlMsgTypeStr[MAX_FLOW_CONTROL_MSG_TYPE] =
 {
 		"REGISTER_REQ",
 		"REGISTER_RESP",
 		"REPORT_REQ",
 		"REPORT_RESP"
 };

 //! Flow Control Msg Type to String
 inline const char *FlowControlMsgTypeToStr(FlowControlMsgType type)
 {
 	if (type < MAX_FLOW_CONTROL_MSG_TYPE)
 		return FlowControlMsgTypeStr[type];
 	else
 		return NULL;
 }

 //! FlowControll Protocol Msg ID (Some Messages are fix length, Some are variable length (TLV)
 typedef enum {
 	//Fix length 8 bytes: client to server in register request, required field
 	FLOW_SERIAL_NUMBER,
 	// Flow XML name TLV: client to server in register request and report request, required field
 	FLOW_XML_NAME,
 	// Flow XML content, TLV: server to client in register respond, option field in case server want to ask client to load xml from server
 	FLOW_XML_CONTENT,
 	// Fix length, 4 bytes Report interval in msec: server to client in register respond, option field
 	REPORT_INTERVAL,
 	// Processor Name TLV:  server to client in report respond, option field in case server want to ask client to update processor property
 	PROCESSOR_NAME,
 	// Processor Property Name TLV: server to client in report respond, option field in case server want to ask client to update processor property
 	PROPERTY_NAME,
 	// Processor Property Value TLV: server to client in report respond, option field in case server want to ask client to update processor property
 	PROPERTY_VALUE,
 	// Report Blob TLV: client to server in report request, option field in case client want to pickyback the report blob in report request to server
 	REPORT_BLOB,
 	MAX_FLOW_MSG_ID
 } FlowControlMsgID;

 //! FlowControl Protocol Msg ID String
 static const char *FlowControlMsgIDStr[MAX_FLOW_MSG_ID] =
 {
 		"FLOW_SERIAL_NUMBER",
 		"FLOW_XML_NAME",
 		"FLOW_XML_CONTENT",
 		"REPORT_INTERVAL",
 		"PROCESSOR_NAME"
 		"PROPERTY_NAME",
 		"PROPERTY_VALUE",
 		"REPORT_BLOB"
 };

 #define TYPE_HDR_LEN 4 // Fix Hdr Type
 #define TLV_HDR_LEN 8 // Type 4 bytes and Len 4 bytes

 //! FlowControl Protocol Msg Len
 inline int FlowControlMsgIDEncodingLen(FlowControlMsgID id, int payLoadLen)
 {
 	if (id == FLOW_SERIAL_NUMBER)
 		return (TYPE_HDR_LEN + 8);
 	else if (id == REPORT_INTERVAL)
 		return (TYPE_HDR_LEN + 4);
 	else if (id < MAX_FLOW_MSG_ID)
 		return (TLV_HDR_LEN + payLoadLen);
 	else
 		return -1;
 }

 //! Flow Control Msg Id to String
 inline const char *FlowControlMsgIdToStr(FlowControlMsgID id)
 {
 	if (id < MAX_FLOW_MSG_ID)
 		return FlowControlMsgIDStr[id];
 	else
 		return NULL;
 }

 //! Flow Control Respond status code
 typedef enum {
 	RESP_SUCCESS,
 	RESP_TRIGGER_REGISTER, // Server respond to client report to re trigger register
 	RESP_START_FLOW_CONTROLLER, // Server respond to client to start flow controller
 	RESP_STOP_FLOW_CONTROLLER, // Server respond to client to stop flow controller
 	RESP_FAILURE,
 	MAX_RESP_CODE
 } FlowControlRespCode;

 //! FlowControl Resp Code str
 static const char *FlowControlRespCodeStr[MAX_RESP_CODE] =
 {
 		"RESP_SUCCESS",
 		"RESP_TRIGGER_REGISTER",
 		"RESP_START_FLOW_CONTROLLER",
 		"RESP_STOP_FLOW_CONTROLLER",
 		"RESP_FAILURE"
 };

 //! Flow Control Resp Code to String
 inline const char *FlowControlRespCodeToStr(FlowControlRespCode code)
 {
 	if (code < MAX_RESP_CODE)
 		return FlowControlRespCodeStr[code];
 	else
 		return NULL;
 }

 //! Common FlowControlProtocol Header
 typedef struct {
 	uint32_t msgType; //! Msg Type
 	uint32_t seqNumber; //! Seq Number to match Req with Resp
 	uint32_t status; //! Resp Code, see FlowControlRespCode
 	uint32_t payloadLen; //! Msg Payload length
 } FlowControlProtocolHeader;

 //! FlowControlProtocol Class
 class FlowControlProtocol
 {
 public:
 	//! Constructor
 	/*!
 	 * Create a new control protocol
 	 */
 	FlowControlProtocol(FlowController *controller) {
 		_controller = controller;
 		_logger = Logger::getLogger();
 		_configure = Configure::getConfigure();
 		_socket = 0;
 		_serverName = "localhost";
 		_serverPort = DEFAULT_NIFI_SERVER_PORT;
 		_registered = false;
 		_seqNumber = 0;
 		_reportBlob = NULL;
 		_reportBlobLen = 0;
 		_reportInterval = DEFAULT_REPORT_INTERVAL;
 		_running = false;

 		std::string value;

 		if (_configure->get(Configure::nifi_server_name, value))
 		{
 			_serverName = value;
 			_logger->log_info("NiFi Server Name %s", _serverName.c_str());
 		}
 		if (_configure->get(Configure::nifi_server_port, value) && Property::StringToInt(value, _serverPort))
 		{
 			_logger->log_info("NiFi Server Port: [%d]", _serverPort);
 		}
 		if (_configure->get(Configure::nifi_server_report_interval, value))
 		{
 			TimeUnit unit;
 			if (Property::StringToTime(value, _reportInterval, unit) &&
 						Property::ConvertTimeUnitToMS(_reportInterval, unit, _reportInterval))
 			{
 				_logger->log_info("NiFi server report interval: [%d] ms", _reportInterval);
 			}
 		}
 	}
 	//! Destructor
 	virtual ~FlowControlProtocol()
 	{
 		stop();
 		if (_socket)
 			close(_socket);
 		if (_reportBlob)
 			delete [] _reportBlob;
 		if (this->_thread)
 			delete this->_thread;
 	}

 public:

 	//! SendRegisterRequest and Process Register Respond, return 0 for success
 	int sendRegisterReq();
 	//! SendReportReq and Process Report Respond, return 0 for success
 	int sendReportReq();
 	//! Start the flow control protocol
 	void start();
 	//! Stop the flow control protocol
 	void stop();
 	//! Set Report BLOB for periodically report
 	void setReportBlob(char *blob, int len)
 	{
 		std::lock_guard<std::mutex> lock(_mtx);
 		if (_reportBlob && _reportBlobLen >= len)
 		{
 			memcpy(_reportBlob, blob, len);
 			_reportBlobLen = len;
 		}
 		else
 		{
 			if (_reportBlob)
 				delete[] _reportBlob;
 			_reportBlob = new char[len];
 			_reportBlobLen = len;
 		}
 	}
 	//! Run function for the thread
 	static void run(FlowControlProtocol *protocol);
 	//! set 8 bytes SerialNumber
 	void setSerialNumber(uint8_t *number)
 	{
 		memcpy(_serialNumber, number, 8);
 	}

 protected:

 private:
 	//! Connect to the socket, return sock descriptor if success, 0 for failure
 	int connectServer(const char *host, uint16_t port);
 	//! Send Data via the socket, return -1 for failure
 	int sendData(uint8_t *buf, int buflen);
 	//! Read length into buf, return -1 for failure and 0 for EOF
 	int readData(uint8_t *buf, int buflen);
 	//! Select on the socket
 	int selectClient(int msec);
 	//! Read the header
 	int readHdr(FlowControlProtocolHeader *hdr);
 	//! encode uint32_t
 	uint8_t *encode(uint8_t *buf, uint32_t value)
 	{
 		*buf++ = (value & 0xFF000000) >> 24;
 		*buf++ = (value & 0x00FF0000) >> 16;
 		*buf++ = (value & 0x0000FF00) >> 8;
 		*buf++ = (value & 0x000000FF);
 		return buf;
 	}
 	//! encode uint32_t
 	uint8_t *decode(uint8_t *buf, uint32_t &value)
 	{
 		value = ((buf[0]<<24)|(buf[1]<<16)|(buf[2]<<8)|(buf[3]));
 		return (buf + 4);
 	}
 	//! encode byte array
 	uint8_t *encode(uint8_t *buf, uint8_t *bufArray, int size)
 	{
 		memcpy(buf, bufArray, size);
 		buf += size;
 		return buf;
 	}
 	//! encode std::string
 	uint8_t *encode(uint8_t *buf, std::string value)
 	{
 		// add the \0 for size
 		buf = encode(buf, value.size()+1);
 		buf = encode(buf, (uint8_t *) value.c_str(), value.size()+1);
 		return buf;
 	}
 	//! Mutex for protection
 	std::mutex _mtx;
 	//! Logger
 	Logger *_logger;
 	//! Configure
 	Configure *_configure;
 	//! NiFi server Name
 	std::string _serverName;
 	//! NiFi server port
 	int64_t _serverPort;
 	//! Serial Number
 	uint8_t _serialNumber[8];
 	//! socket to server
 	int _socket;
 	//! report interal in msec
 	int64_t _reportInterval;
 	//! whether it was registered to the NiFi server
 	bool _registered;
 	//! seq number
 	uint32_t _seqNumber;
 	//! FlowController
 	FlowController *_controller;
 	//! report Blob
 	char *_reportBlob;
 	//! report Blob len;
 	int _reportBlobLen;
 	//! thread
 	std::thread *_thread;
 	//! whether it is running
 	bool _running;
 	// Prevent default copy constructor and assignment operation
 	// Only support pass by reference or pointer
 	FlowControlProtocol(const FlowControlProtocol &parent);
 	FlowControlProtocol &operator=(const FlowControlProtocol &parent);

 };

 #endif
	/**
	* @file FlowControlProtocol.h
	* FlowControlProtocol 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 __FLOW_CONTROL_PROTOCOL_H__
	#define __FLOW_CONTROL_PROTOCOL_H__

	#include <stdio.h>
	#include <unistd.h>
	#include <sys/types.h>
	#include <sys/socket.h>
	#include <netinet/in.h>
	#include <arpa/inet.h>
	#include <fcntl.h>
	#include <netdb.h>
	#include <string>
	#include <errno.h>
	#include <chrono>
	#include <thread>
	#include "Logger.h"
	#include "Configure.h"
	#include "Property.h"

	//! Forwarder declaration
	class FlowController;

	#define DEFAULT_NIFI_SERVER_PORT 9000
	#define DEFAULT_REPORT_INTERVAL 1000 // 1 sec
	#define MAX_READ_TIMEOUT 30000 // 30 seconds

	//! FlowControl Protocol Msg Type
	typedef enum {
	REGISTER_REQ, // Device Register Request from device to server which contain device serial number, current running flow xml version
	REGISTER_RESP, // Device Register Respond from server to device, may contain new flow.xml from server ask device to apply and also device report interval
	REPORT_REQ, // Period Device Report from device to server which contain device serial number, current running flow xml name/version and other period report info
	REPORT_RESP, // Report Respond from server to device, may ask device to update flow xml or processor property
	MAX_FLOW_CONTROL_MSG_TYPE
	} FlowControlMsgType;

	//! FlowControl Protocol Msg Type String
	static const char *FlowControlMsgTypeStr[MAX_FLOW_CONTROL_MSG_TYPE] =
	{
	"REGISTER_REQ",
	"REGISTER_RESP",
	"REPORT_REQ",
	"REPORT_RESP"
	};

	//! Flow Control Msg Type to String
	inline const char *FlowControlMsgTypeToStr(FlowControlMsgType type)
	{
	if (type < MAX_FLOW_CONTROL_MSG_TYPE)
	return FlowControlMsgTypeStr[type];
	else
	return NULL;
	}

	//! FlowControll Protocol Msg ID (Some Messages are fix length, Some are variable length (TLV)
	typedef enum {
	//Fix length 8 bytes: client to server in register request, required field
	FLOW_SERIAL_NUMBER,
	// Flow XML name TLV: client to server in register request and report request, required field
	FLOW_XML_NAME,
	// Flow XML content, TLV: server to client in register respond, option field in case server want to ask client to load xml from server
	FLOW_XML_CONTENT,
	// Fix length, 4 bytes Report interval in msec: server to client in register respond, option field
	REPORT_INTERVAL,
	// Processor Name TLV: server to client in report respond, option field in case server want to ask client to update processor property
	PROCESSOR_NAME,
	// Processor Property Name TLV: server to client in report respond, option field in case server want to ask client to update processor property
	PROPERTY_NAME,
	// Processor Property Value TLV: server to client in report respond, option field in case server want to ask client to update processor property
	PROPERTY_VALUE,
	// Report Blob TLV: client to server in report request, option field in case client want to pickyback the report blob in report request to server
	REPORT_BLOB,
	MAX_FLOW_MSG_ID
	} FlowControlMsgID;

	//! FlowControl Protocol Msg ID String
	static const char *FlowControlMsgIDStr[MAX_FLOW_MSG_ID] =
	{
	"FLOW_SERIAL_NUMBER",
	"FLOW_XML_NAME",
	"FLOW_XML_CONTENT",
	"REPORT_INTERVAL",
	"PROCESSOR_NAME"
	"PROPERTY_NAME",
	"PROPERTY_VALUE",
	"REPORT_BLOB"
	};

	#define TYPE_HDR_LEN 4 // Fix Hdr Type
	#define TLV_HDR_LEN 8 // Type 4 bytes and Len 4 bytes

	//! FlowControl Protocol Msg Len
	inline int FlowControlMsgIDEncodingLen(FlowControlMsgID id, int payLoadLen)
	{
	if (id == FLOW_SERIAL_NUMBER)
	return (TYPE_HDR_LEN + 8);
	else if (id == REPORT_INTERVAL)
	return (TYPE_HDR_LEN + 4);
	else if (id < MAX_FLOW_MSG_ID)
	return (TLV_HDR_LEN + payLoadLen);
	else
	return -1;
	}

	//! Flow Control Msg Id to String
	inline const char *FlowControlMsgIdToStr(FlowControlMsgID id)
	{
	if (id < MAX_FLOW_MSG_ID)
	return FlowControlMsgIDStr[id];
	else
	return NULL;
	}

	//! Flow Control Respond status code
	typedef enum {
	RESP_SUCCESS,
	RESP_TRIGGER_REGISTER, // Server respond to client report to re trigger register
	RESP_START_FLOW_CONTROLLER, // Server respond to client to start flow controller
	RESP_STOP_FLOW_CONTROLLER, // Server respond to client to stop flow controller
	RESP_FAILURE,
	MAX_RESP_CODE
	} FlowControlRespCode;

	//! FlowControl Resp Code str
	static const char *FlowControlRespCodeStr[MAX_RESP_CODE] =
	{
	"RESP_SUCCESS",
	"RESP_TRIGGER_REGISTER",
	"RESP_START_FLOW_CONTROLLER",
	"RESP_STOP_FLOW_CONTROLLER",
	"RESP_FAILURE"
	};

	//! Flow Control Resp Code to String
	inline const char *FlowControlRespCodeToStr(FlowControlRespCode code)
	{
	if (code < MAX_RESP_CODE)
	return FlowControlRespCodeStr[code];
	else
	return NULL;
	}

	//! Common FlowControlProtocol Header
	typedef struct {
	uint32_t msgType; //! Msg Type
	uint32_t seqNumber; //! Seq Number to match Req with Resp
	uint32_t status; //! Resp Code, see FlowControlRespCode
	uint32_t payloadLen; //! Msg Payload length
	} FlowControlProtocolHeader;

	//! FlowControlProtocol Class
	class FlowControlProtocol
	{
	public:
	//! Constructor
	/*!
	* Create a new control protocol
	*/
	FlowControlProtocol(FlowController *controller) {
	_controller = controller;
	_logger = Logger::getLogger();
	_configure = Configure::getConfigure();
	_socket = 0;
	_serverName = "localhost";
	_serverPort = DEFAULT_NIFI_SERVER_PORT;
	_registered = false;
	_seqNumber = 0;
	_reportBlob = NULL;
	_reportBlobLen = 0;
	_reportInterval = DEFAULT_REPORT_INTERVAL;
	_running = false;

	std::string value;

	if (_configure->get(Configure::nifi_server_name, value))
	{
	_serverName = value;
	_logger->log_info("NiFi Server Name %s", _serverName.c_str());
	}
	if (_configure->get(Configure::nifi_server_port, value) && Property::StringToInt(value, _serverPort))
	{
	_logger->log_info("NiFi Server Port: [%d]", _serverPort);
	}
	if (_configure->get(Configure::nifi_server_report_interval, value))
	{
	TimeUnit unit;
	if (Property::StringToTime(value, _reportInterval, unit) &&
	Property::ConvertTimeUnitToMS(_reportInterval, unit, _reportInterval))
	{
	_logger->log_info("NiFi server report interval: [%d] ms", _reportInterval);
	}
	}
	}
	//! Destructor
	virtual ~FlowControlProtocol()
	{
	stop();
	if (_socket)
	close(_socket);
	if (_reportBlob)
	delete [] _reportBlob;
	if (this->_thread)
	delete this->_thread;
	}

	public:

	//! SendRegisterRequest and Process Register Respond, return 0 for success
	int sendRegisterReq();
	//! SendReportReq and Process Report Respond, return 0 for success
	int sendReportReq();
	//! Start the flow control protocol
	void start();
	//! Stop the flow control protocol
	void stop();
	//! Set Report BLOB for periodically report
	void setReportBlob(char *blob, int len)
	{
	std::lock_guard<std::mutex> lock(_mtx);
	if (_reportBlob && _reportBlobLen >= len)
	{
	memcpy(_reportBlob, blob, len);
	_reportBlobLen = len;
	}
	else
	{
	if (_reportBlob)
	delete[] _reportBlob;
	_reportBlob = new char[len];
	_reportBlobLen = len;
	}
	}
	//! Run function for the thread
	static void run(FlowControlProtocol *protocol);
	//! set 8 bytes SerialNumber
	void setSerialNumber(uint8_t *number)
	{
	memcpy(_serialNumber, number, 8);
	}

	protected:

	private:
	//! Connect to the socket, return sock descriptor if success, 0 for failure
	int connectServer(const char *host, uint16_t port);
	//! Send Data via the socket, return -1 for failure
	int sendData(uint8_t *buf, int buflen);
	//! Read length into buf, return -1 for failure and 0 for EOF
	int readData(uint8_t *buf, int buflen);
	//! Select on the socket
	int selectClient(int msec);
	//! Read the header
	int readHdr(FlowControlProtocolHeader *hdr);
	//! encode uint32_t
	uint8_t encode(uint8_t buf, uint32_t value)
	{
	*buf++ = (value & 0xFF000000) >> 24;
	*buf++ = (value & 0x00FF0000) >> 16;
	*buf++ = (value & 0x0000FF00) >> 8;
	*buf++ = (value & 0x000000FF);
	return buf;
	}
	//! encode uint32_t
	uint8_t decode(uint8_t buf, uint32_t &value)
	{
	value = ((buf[0]<<24)\|(buf[1]<<16)\|(buf[2]<<8)\|(buf[3]));
	return (buf + 4);
	}
	//! encode byte array
	uint8_t encode(uint8_t buf, uint8_t *bufArray, int size)
	{
	memcpy(buf, bufArray, size);
	buf += size;
	return buf;
	}
	//! encode std::string
	uint8_t encode(uint8_t buf, std::string value)
	{
	// add the \0 for size
	buf = encode(buf, value.size()+1);
	buf = encode(buf, (uint8_t *) value.c_str(), value.size()+1);
	return buf;
	}
	//! Mutex for protection
	std::mutex _mtx;
	//! Logger
	Logger *_logger;
	//! Configure
	Configure *_configure;
	//! NiFi server Name
	std::string _serverName;
	//! NiFi server port
	int64_t _serverPort;
	//! Serial Number
	uint8_t _serialNumber[8];
	//! socket to server
	int _socket;
	//! report interal in msec
	int64_t _reportInterval;
	//! whether it was registered to the NiFi server
	bool _registered;
	//! seq number
	uint32_t _seqNumber;
	//! FlowController
	FlowController *_controller;
	//! report Blob
	char *_reportBlob;
	//! report Blob len;
	int _reportBlobLen;
	//! thread
	std::thread *_thread;
	//! whether it is running
	bool _running;
	// Prevent default copy constructor and assignment operation
	// Only support pass by reference or pointer
	FlowControlProtocol(const FlowControlProtocol &parent);
	FlowControlProtocol &operator=(const FlowControlProtocol &parent);

	};

	#endif