libminifi/src/FlowControlProtocol.cpp - nifi-minifi-cpp - Git at Google

 /**
  * @file FlowControlProtocol.cpp
  * FlowControlProtocol class implementation
  *
  * 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.
  */
 #include "FlowControlProtocol.h"
 #include <sys/time.h>
 #include <stdio.h>
 #include <time.h>
 #include <netinet/tcp.h>
 #include <chrono>
 #include <thread>
 #include <string>
 #include <random>
 #include <iostream>
 #include "FlowController.h"
 #include "core/Core.h"
 namespace org {
 namespace apache {
 namespace nifi {
 namespace minifi {

 int FlowControlProtocol::connectServer(const char *host, uint16_t port) {
   in_addr_t addr;
   int sock = 0;
   struct hostent *h;
 #ifdef __MACH__
   h = gethostbyname(host);
 #else
   char buf[1024];
   struct hostent he;
   int hh_errno;
   gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno);
 #endif
   memcpy(reinterpret_cast<char*>(&addr), h->h_addr_list[0], h->h_length);
   sock = socket(AF_INET, SOCK_STREAM, 0);
   if (sock < 0) {
     logger_->log_error("Could not create socket to hostName %s", host);
     return 0;
   }

 #ifndef __MACH__
   int opt = 1;
   bool nagle_off = true;

   if (nagle_off) {
     if (setsockopt(sock, SOL_TCP, TCP_NODELAY, reinterpret_cast<void*>(&opt), sizeof(opt)) < 0) {
       logger_->log_error("setsockopt() TCP_NODELAY failed");
       close(sock);
       return 0;
     }
     if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char*>(&opt), sizeof(opt)) < 0) {
       logger_->log_error("setsockopt() SO_REUSEADDR failed");
       close(sock);
       return 0;
     }
   }

   int sndsize = 256 * 1024;
   if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, reinterpret_cast<char*>(&sndsize), sizeof(sndsize)) < 0) {
     logger_->log_error("setsockopt() SO_SNDBUF failed");
     close(sock);
     return 0;
   }
 #endif

   struct sockaddr_in sa;
   socklen_t socklen;
   int status;

   memset(&sa, 0, sizeof(sa));
   sa.sin_family = AF_INET;
   sa.sin_addr.s_addr = htonl(INADDR_ANY);
   sa.sin_port = htons(0);
   socklen = sizeof(sa);
   if (bind(sock, (struct sockaddr *) &sa, socklen) < 0) {
     logger_->log_error("socket bind failed");
     close(sock);
     return 0;
   }

   memset(&sa, 0, sizeof(sa));
   sa.sin_family = AF_INET;
   sa.sin_addr.s_addr = addr;
   sa.sin_port = htons(port);
   socklen = sizeof(sa);

   status = connect(sock, (struct sockaddr *) &sa, socklen);

   if (status < 0) {
     logger_->log_error("socket connect failed to %s %ll", host, port);
     close(sock);
     return 0;
   }

   logger_->log_debug("Flow Control Protocol socket %ll connect to server %s port %ll success", sock, host, port);

   return sock;
 }

 int FlowControlProtocol::sendData(uint8_t *buf, int buflen) {
   int ret = 0, bytes = 0;

   while (bytes < buflen) {
     ret = send(_socket, buf + bytes, buflen - bytes, 0);
     // check for errors
     if (ret == -1) {
       return ret;
     }
     bytes += ret;
   }

   return bytes;
 }

 int FlowControlProtocol::selectClient(int msec) {
   fd_set fds;
   struct timeval tv;
   int retval;
   int fd = _socket;

   FD_ZERO(&fds);
   FD_SET(fd, &fds);

   tv.tv_sec = msec / 1000;
   tv.tv_usec = (msec % 1000) * 1000;

   if (msec > 0)
     retval = select(fd + 1, &fds, NULL, NULL, &tv);
   else
     retval = select(fd + 1, &fds, NULL, NULL, NULL);

   if (retval <= 0)
     return retval;
   if (FD_ISSET(fd, &fds))
     return retval;
   else
     return 0;
 }

 int FlowControlProtocol::readData(uint8_t *buf, int buflen) {
   int sendSize = buflen;

   while (buflen) {
     int status;
     status = selectClient(MAX_READ_TIMEOUT);
     if (status <= 0) {
       return status;
     }
 #ifndef __MACH__
     status = read(_socket, buf, buflen);
 #else
     status = recv(_socket, buf, buflen, 0);
 #endif
     if (status <= 0) {
       return status;
     }
     buflen -= status;
     buf += status;
   }

   return sendSize;
 }

 int FlowControlProtocol::readHdr(FlowControlProtocolHeader *hdr) {
   uint8_t buffer[sizeof(FlowControlProtocolHeader)];

   uint8_t *data = buffer;

   int status = readData(buffer, sizeof(FlowControlProtocolHeader));
   if (status <= 0)
     return status;

   uint32_t value;
   data = this->decode(data, value);
   hdr->msgType = value;

   data = this->decode(data, value);
   hdr->seqNumber = value;

   data = this->decode(data, value);
   hdr->status = value;

   data = this->decode(data, value);
   hdr->payloadLen = value;

   return sizeof(FlowControlProtocolHeader);
 }

 void FlowControlProtocol::start() {
   if (_reportInterval <= 0)
     return;
   if (running_)
     return;
   running_ = true;
   logger_->log_trace("FlowControl Protocol Start");
   _thread = new std::thread(run, this);
   _thread->detach();
 }

 void FlowControlProtocol::stop() {
   if (!running_)
     return;
   running_ = false;
   logger_->log_info("FlowControl Protocol Stop");
 }

 void FlowControlProtocol::run(FlowControlProtocol *protocol) {
   while (protocol->running_) {
     std::this_thread::sleep_for(std::chrono::milliseconds(protocol->_reportInterval));
     if (!protocol->_registered) {
       // if it is not register yet
       protocol->sendRegisterReq();
     } else {
       protocol->sendReportReq();
     }
   }
   return;
 }

 int FlowControlProtocol::sendRegisterReq() {
   if (_registered) {
     logger_->log_debug("Already registered");
     return -1;
   }

   uint16_t port = this->_serverPort;

   if (this->_socket <= 0)
     this->_socket = connectServer(_serverName.c_str(), port);

   if (this->_socket <= 0)
     return -1;

   // Calculate the total payload msg size
   uint32_t payloadSize = FlowControlMsgIDEncodingLen(FLOW_SERIAL_NUMBER, 0) + FlowControlMsgIDEncodingLen(FLOW_YML_NAME, this->_controller->getName().size() + 1);
   uint32_t size = sizeof(FlowControlProtocolHeader) + payloadSize;

   uint8_t *data = new uint8_t[size];
   uint8_t *start = data;

   // encode the HDR
   FlowControlProtocolHeader hdr;
   hdr.msgType = REGISTER_REQ;
   hdr.payloadLen = payloadSize;
   hdr.seqNumber = this->_seqNumber;
   hdr.status = RESP_SUCCESS;
   data = this->encode(data, hdr.msgType);
   data = this->encode(data, hdr.seqNumber);
   data = this->encode(data, hdr.status);
   data = this->encode(data, hdr.payloadLen);

   // encode the serial number
   data = this->encode(data, FLOW_SERIAL_NUMBER);
   data = this->encode(data, this->_serialNumber, 8);

   // encode the YAML name
   data = this->encode(data, FLOW_YML_NAME);
   data = this->encode(data, this->_controller->getName());

   // send it
   int status = sendData(start, size);
   delete[] start;
   if (status <= 0) {
     close(_socket);
     _socket = 0;
     logger_->log_error("Flow Control Protocol Send Register Req failed");
     return -1;
   }

   // Looking for register respond
   status = readHdr(&hdr);

   if (status <= 0) {
     close(_socket);
     _socket = 0;
     logger_->log_error("Flow Control Protocol Read Register Resp header failed");
     return -1;
   }
   logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
   logger_->log_debug("Flow Control Protocol receive Seq Num %ll", hdr.seqNumber);
   logger_->log_debug("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status));
   logger_->log_debug("Flow Control Protocol receive Payload len %ll", hdr.payloadLen);

   if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) {
     this->_registered = true;
     this->_seqNumber++;
     logger_->log_trace("Flow Control Protocol Register success");
     uint8_t *payload = new uint8_t[hdr.payloadLen];
     uint8_t *payloadPtr = payload;
     status = readData(payload, hdr.payloadLen);
     if (status <= 0) {
       delete[] payload;
       logger_->log_warn("Flow Control Protocol Register Read Payload fail");
       close(_socket);
       _socket = 0;
       return -1;
     }
     while (payloadPtr < (payload + hdr.payloadLen)) {
       uint32_t msgID;
       payloadPtr = this->decode(payloadPtr, msgID);
       if (((FlowControlMsgID) msgID) == REPORT_INTERVAL) {
         // Fixed 4 bytes
         uint32_t reportInterval;
         payloadPtr = this->decode(payloadPtr, reportInterval);
         logger_->log_debug("Flow Control Protocol receive report interval %ll ms", reportInterval);
         this->_reportInterval = reportInterval;
       } else {
         break;
       }
     }
     delete[] payload;
     close(_socket);
     _socket = 0;
     return 0;
   } else {
     logger_->log_warn("Flow Control Protocol Register fail");
     close(_socket);
     _socket = 0;
     return -1;
   }
 }

 int FlowControlProtocol::sendReportReq() {
   uint16_t port = this->_serverPort;

   if (this->_socket <= 0)
     this->_socket = connectServer(_serverName.c_str(), port);

   if (this->_socket <= 0)
     return -1;

   // Calculate the total payload msg size
   uint32_t payloadSize = FlowControlMsgIDEncodingLen(FLOW_YML_NAME, this->_controller->getName().size() + 1);
   uint32_t size = sizeof(FlowControlProtocolHeader) + payloadSize;

   uint8_t *data = new uint8_t[size];
   uint8_t *start = data;

   // encode the HDR
   FlowControlProtocolHeader hdr;
   hdr.msgType = REPORT_REQ;
   hdr.payloadLen = payloadSize;
   hdr.seqNumber = this->_seqNumber;
   hdr.status = RESP_SUCCESS;
   data = this->encode(data, hdr.msgType);
   data = this->encode(data, hdr.seqNumber);
   data = this->encode(data, hdr.status);
   data = this->encode(data, hdr.payloadLen);

   // encode the YAML name
   data = this->encode(data, FLOW_YML_NAME);
   data = this->encode(data, this->_controller->getName());

   // send it
   int status = sendData(start, size);
   delete[] start;
   if (status <= 0) {
     close(_socket);
     _socket = 0;
     logger_->log_error("Flow Control Protocol Send Report Req failed");
     return -1;
   }

   // Looking for report respond
   status = readHdr(&hdr);

   if (status <= 0) {
     close(_socket);
     _socket = 0;
     logger_->log_error("Flow Control Protocol Read Report Resp header failed");
     return -1;
   }
   logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
   logger_->log_debug("Flow Control Protocol receive Seq Num %ll", hdr.seqNumber);
   logger_->log_debug("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status));
   logger_->log_debug("Flow Control Protocol receive Payload len %ll", hdr.payloadLen);

   if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) {
     this->_seqNumber++;
     uint8_t *payload = new uint8_t[hdr.payloadLen];
     uint8_t *payloadPtr = payload;
     status = readData(payload, hdr.payloadLen);
     if (status <= 0) {
       delete[] payload;
       logger_->log_warn("Flow Control Protocol Report Resp Read Payload fail");
       close(_socket);
       _socket = 0;
       return -1;
     }
     std::string processor;
     std::string propertyName;
     std::string propertyValue;
     while (payloadPtr < (payload + hdr.payloadLen)) {
       uint32_t msgID;
       payloadPtr = this->decode(payloadPtr, msgID);
       if (((FlowControlMsgID) msgID) == PROCESSOR_NAME) {
         uint32_t len;
         payloadPtr = this->decode(payloadPtr, len);
         processor = (const char *) payloadPtr;
         payloadPtr += len;
         logger_->log_debug("Flow Control Protocol receive report resp processor %s", processor);
       } else if (((FlowControlMsgID) msgID) == PROPERTY_NAME) {
         uint32_t len;
         payloadPtr = this->decode(payloadPtr, len);
         propertyName = (const char *) payloadPtr;
         payloadPtr += len;
         logger_->log_debug("Flow Control Protocol receive report resp property name %s", propertyName);
       } else if (((FlowControlMsgID) msgID) == PROPERTY_VALUE) {
         uint32_t len;
         payloadPtr = this->decode(payloadPtr, len);
         propertyValue = (const char *) payloadPtr;
         payloadPtr += len;
         logger_->log_debug("Flow Control Protocol receive report resp property value %s", propertyValue);
         this->_controller->updatePropertyValue(processor, propertyName, propertyValue);
       } else {
         break;
       }
     }
     delete[] payload;
     close(_socket);
     _socket = 0;
     return 0;
   } else if (hdr.status == RESP_TRIGGER_REGISTER && hdr.seqNumber == this->_seqNumber) {
     logger_->log_trace("Flow Control Protocol trigger reregister");
     this->_registered = false;
     this->_seqNumber++;
     close(_socket);
     _socket = 0;
     return 0;
   } else if (hdr.status == RESP_STOP_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) {
     logger_->log_trace("Flow Control Protocol stop flow controller");
     this->_controller->stop(true);
     this->_seqNumber++;
     close(_socket);
     _socket = 0;
     return 0;
   } else if (hdr.status == RESP_START_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) {
     logger_->log_trace("Flow Control Protocol start flow controller");
     this->_controller->start();
     this->_seqNumber++;
     close(_socket);
     _socket = 0;
     return 0;
   } else {
     logger_->log_trace("Flow Control Protocol Report fail");
     close(_socket);
     _socket = 0;
     return -1;
   }
 }

 } /* namespace minifi */
 } /* namespace nifi */
 } /* namespace apache */
 } /* namespace org */
	/**
	* @file FlowControlProtocol.cpp
	* FlowControlProtocol class implementation
	*
	* 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.
	*/
	#include "FlowControlProtocol.h"
	#include <sys/time.h>
	#include <stdio.h>
	#include <time.h>
	#include <netinet/tcp.h>
	#include <chrono>
	#include <thread>
	#include <string>
	#include <random>
	#include <iostream>
	#include "FlowController.h"
	#include "core/Core.h"
	namespace org {
	namespace apache {
	namespace nifi {
	namespace minifi {

	int FlowControlProtocol::connectServer(const char *host, uint16_t port) {
	in_addr_t addr;
	int sock = 0;
	struct hostent *h;
	#ifdef __MACH__
	h = gethostbyname(host);
	#else
	char buf[1024];
	struct hostent he;
	int hh_errno;
	gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno);
	#endif
	memcpy(reinterpret_cast<char*>(&addr), h->h_addr_list[0], h->h_length);
	sock = socket(AF_INET, SOCK_STREAM, 0);
	if (sock < 0) {
	logger_->log_error("Could not create socket to hostName %s", host);
	return 0;
	}

	#ifndef __MACH__
	int opt = 1;
	bool nagle_off = true;

	if (nagle_off) {
	if (setsockopt(sock, SOL_TCP, TCP_NODELAY, reinterpret_cast<void*>(&opt), sizeof(opt)) < 0) {
	logger_->log_error("setsockopt() TCP_NODELAY failed");
	close(sock);
	return 0;
	}
	if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char*>(&opt), sizeof(opt)) < 0) {
	logger_->log_error("setsockopt() SO_REUSEADDR failed");
	close(sock);
	return 0;
	}
	}

	int sndsize = 256 * 1024;
	if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, reinterpret_cast<char*>(&sndsize), sizeof(sndsize)) < 0) {
	logger_->log_error("setsockopt() SO_SNDBUF failed");
	close(sock);
	return 0;
	}
	#endif

	struct sockaddr_in sa;
	socklen_t socklen;
	int status;

	memset(&sa, 0, sizeof(sa));
	sa.sin_family = AF_INET;
	sa.sin_addr.s_addr = htonl(INADDR_ANY);
	sa.sin_port = htons(0);
	socklen = sizeof(sa);
	if (bind(sock, (struct sockaddr *) &sa, socklen) < 0) {
	logger_->log_error("socket bind failed");
	close(sock);
	return 0;
	}

	memset(&sa, 0, sizeof(sa));
	sa.sin_family = AF_INET;
	sa.sin_addr.s_addr = addr;
	sa.sin_port = htons(port);
	socklen = sizeof(sa);

	status = connect(sock, (struct sockaddr *) &sa, socklen);

	if (status < 0) {
	logger_->log_error("socket connect failed to %s %ll", host, port);
	close(sock);
	return 0;
	}

	logger_->log_debug("Flow Control Protocol socket %ll connect to server %s port %ll success", sock, host, port);

	return sock;
	}

	int FlowControlProtocol::sendData(uint8_t *buf, int buflen) {
	int ret = 0, bytes = 0;

	while (bytes < buflen) {
	ret = send(_socket, buf + bytes, buflen - bytes, 0);
	// check for errors
	if (ret == -1) {
	return ret;
	}
	bytes += ret;
	}

	return bytes;
	}

	int FlowControlProtocol::selectClient(int msec) {
	fd_set fds;
	struct timeval tv;
	int retval;
	int fd = _socket;

	FD_ZERO(&fds);
	FD_SET(fd, &fds);

	tv.tv_sec = msec / 1000;
	tv.tv_usec = (msec % 1000) * 1000;

	if (msec > 0)
	retval = select(fd + 1, &fds, NULL, NULL, &tv);
	else
	retval = select(fd + 1, &fds, NULL, NULL, NULL);

	if (retval <= 0)
	return retval;
	if (FD_ISSET(fd, &fds))
	return retval;
	else
	return 0;
	}

	int FlowControlProtocol::readData(uint8_t *buf, int buflen) {
	int sendSize = buflen;

	while (buflen) {
	int status;
	status = selectClient(MAX_READ_TIMEOUT);
	if (status <= 0) {
	return status;
	}
	#ifndef __MACH__
	status = read(_socket, buf, buflen);
	#else
	status = recv(_socket, buf, buflen, 0);
	#endif
	if (status <= 0) {
	return status;
	}
	buflen -= status;
	buf += status;
	}

	return sendSize;
	}

	int FlowControlProtocol::readHdr(FlowControlProtocolHeader *hdr) {
	uint8_t buffer[sizeof(FlowControlProtocolHeader)];

	uint8_t *data = buffer;

	int status = readData(buffer, sizeof(FlowControlProtocolHeader));
	if (status <= 0)
	return status;

	uint32_t value;
	data = this->decode(data, value);
	hdr->msgType = value;

	data = this->decode(data, value);
	hdr->seqNumber = value;

	data = this->decode(data, value);
	hdr->status = value;

	data = this->decode(data, value);
	hdr->payloadLen = value;

	return sizeof(FlowControlProtocolHeader);
	}

	void FlowControlProtocol::start() {
	if (_reportInterval <= 0)
	return;
	if (running_)
	return;
	running_ = true;
	logger_->log_trace("FlowControl Protocol Start");
	_thread = new std::thread(run, this);
	_thread->detach();
	}

	void FlowControlProtocol::stop() {
	if (!running_)
	return;
	running_ = false;
	logger_->log_info("FlowControl Protocol Stop");
	}

	void FlowControlProtocol::run(FlowControlProtocol *protocol) {
	while (protocol->running_) {
	std::this_thread::sleep_for(std::chrono::milliseconds(protocol->_reportInterval));
	if (!protocol->_registered) {
	// if it is not register yet
	protocol->sendRegisterReq();
	} else {
	protocol->sendReportReq();
	}
	}
	return;
	}

	int FlowControlProtocol::sendRegisterReq() {
	if (_registered) {
	logger_->log_debug("Already registered");
	return -1;
	}

	uint16_t port = this->_serverPort;

	if (this->_socket <= 0)
	this->_socket = connectServer(_serverName.c_str(), port);

	if (this->_socket <= 0)
	return -1;

	// Calculate the total payload msg size
	uint32_t payloadSize = FlowControlMsgIDEncodingLen(FLOW_SERIAL_NUMBER, 0) + FlowControlMsgIDEncodingLen(FLOW_YML_NAME, this->_controller->getName().size() + 1);
	uint32_t size = sizeof(FlowControlProtocolHeader) + payloadSize;

	uint8_t *data = new uint8_t[size];
	uint8_t *start = data;

	// encode the HDR
	FlowControlProtocolHeader hdr;
	hdr.msgType = REGISTER_REQ;
	hdr.payloadLen = payloadSize;
	hdr.seqNumber = this->_seqNumber;
	hdr.status = RESP_SUCCESS;
	data = this->encode(data, hdr.msgType);
	data = this->encode(data, hdr.seqNumber);
	data = this->encode(data, hdr.status);
	data = this->encode(data, hdr.payloadLen);

	// encode the serial number
	data = this->encode(data, FLOW_SERIAL_NUMBER);
	data = this->encode(data, this->_serialNumber, 8);

	// encode the YAML name
	data = this->encode(data, FLOW_YML_NAME);
	data = this->encode(data, this->_controller->getName());

	// send it
	int status = sendData(start, size);
	delete[] start;
	if (status <= 0) {
	close(_socket);
	_socket = 0;
	logger_->log_error("Flow Control Protocol Send Register Req failed");
	return -1;
	}

	// Looking for register respond
	status = readHdr(&hdr);

	if (status <= 0) {
	close(_socket);
	_socket = 0;
	logger_->log_error("Flow Control Protocol Read Register Resp header failed");
	return -1;
	}
	logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
	logger_->log_debug("Flow Control Protocol receive Seq Num %ll", hdr.seqNumber);
	logger_->log_debug("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status));
	logger_->log_debug("Flow Control Protocol receive Payload len %ll", hdr.payloadLen);

	if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) {
	this->_registered = true;
	this->_seqNumber++;
	logger_->log_trace("Flow Control Protocol Register success");
	uint8_t *payload = new uint8_t[hdr.payloadLen];
	uint8_t *payloadPtr = payload;
	status = readData(payload, hdr.payloadLen);
	if (status <= 0) {
	delete[] payload;
	logger_->log_warn("Flow Control Protocol Register Read Payload fail");
	close(_socket);
	_socket = 0;
	return -1;
	}
	while (payloadPtr < (payload + hdr.payloadLen)) {
	uint32_t msgID;
	payloadPtr = this->decode(payloadPtr, msgID);
	if (((FlowControlMsgID) msgID) == REPORT_INTERVAL) {
	// Fixed 4 bytes
	uint32_t reportInterval;
	payloadPtr = this->decode(payloadPtr, reportInterval);
	logger_->log_debug("Flow Control Protocol receive report interval %ll ms", reportInterval);
	this->_reportInterval = reportInterval;
	} else {
	break;
	}
	}
	delete[] payload;
	close(_socket);
	_socket = 0;
	return 0;
	} else {
	logger_->log_warn("Flow Control Protocol Register fail");
	close(_socket);
	_socket = 0;
	return -1;
	}
	}

	int FlowControlProtocol::sendReportReq() {
	uint16_t port = this->_serverPort;

	if (this->_socket <= 0)
	this->_socket = connectServer(_serverName.c_str(), port);

	if (this->_socket <= 0)
	return -1;

	// Calculate the total payload msg size
	uint32_t payloadSize = FlowControlMsgIDEncodingLen(FLOW_YML_NAME, this->_controller->getName().size() + 1);
	uint32_t size = sizeof(FlowControlProtocolHeader) + payloadSize;

	uint8_t *data = new uint8_t[size];
	uint8_t *start = data;

	// encode the HDR
	FlowControlProtocolHeader hdr;
	hdr.msgType = REPORT_REQ;
	hdr.payloadLen = payloadSize;
	hdr.seqNumber = this->_seqNumber;
	hdr.status = RESP_SUCCESS;
	data = this->encode(data, hdr.msgType);
	data = this->encode(data, hdr.seqNumber);
	data = this->encode(data, hdr.status);
	data = this->encode(data, hdr.payloadLen);

	// encode the YAML name
	data = this->encode(data, FLOW_YML_NAME);
	data = this->encode(data, this->_controller->getName());

	// send it
	int status = sendData(start, size);
	delete[] start;
	if (status <= 0) {
	close(_socket);
	_socket = 0;
	logger_->log_error("Flow Control Protocol Send Report Req failed");
	return -1;
	}

	// Looking for report respond
	status = readHdr(&hdr);

	if (status <= 0) {
	close(_socket);
	_socket = 0;
	logger_->log_error("Flow Control Protocol Read Report Resp header failed");
	return -1;
	}
	logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
	logger_->log_debug("Flow Control Protocol receive Seq Num %ll", hdr.seqNumber);
	logger_->log_debug("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status));
	logger_->log_debug("Flow Control Protocol receive Payload len %ll", hdr.payloadLen);

	if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) {
	this->_seqNumber++;
	uint8_t *payload = new uint8_t[hdr.payloadLen];
	uint8_t *payloadPtr = payload;
	status = readData(payload, hdr.payloadLen);
	if (status <= 0) {
	delete[] payload;
	logger_->log_warn("Flow Control Protocol Report Resp Read Payload fail");
	close(_socket);
	_socket = 0;
	return -1;
	}
	std::string processor;
	std::string propertyName;
	std::string propertyValue;
	while (payloadPtr < (payload + hdr.payloadLen)) {
	uint32_t msgID;
	payloadPtr = this->decode(payloadPtr, msgID);
	if (((FlowControlMsgID) msgID) == PROCESSOR_NAME) {
	uint32_t len;
	payloadPtr = this->decode(payloadPtr, len);
	processor = (const char *) payloadPtr;
	payloadPtr += len;
	logger_->log_debug("Flow Control Protocol receive report resp processor %s", processor);
	} else if (((FlowControlMsgID) msgID) == PROPERTY_NAME) {
	uint32_t len;
	payloadPtr = this->decode(payloadPtr, len);
	propertyName = (const char *) payloadPtr;
	payloadPtr += len;
	logger_->log_debug("Flow Control Protocol receive report resp property name %s", propertyName);
	} else if (((FlowControlMsgID) msgID) == PROPERTY_VALUE) {
	uint32_t len;
	payloadPtr = this->decode(payloadPtr, len);
	propertyValue = (const char *) payloadPtr;
	payloadPtr += len;
	logger_->log_debug("Flow Control Protocol receive report resp property value %s", propertyValue);
	this->_controller->updatePropertyValue(processor, propertyName, propertyValue);
	} else {
	break;
	}
	}
	delete[] payload;
	close(_socket);
	_socket = 0;
	return 0;
	} else if (hdr.status == RESP_TRIGGER_REGISTER && hdr.seqNumber == this->_seqNumber) {
	logger_->log_trace("Flow Control Protocol trigger reregister");
	this->_registered = false;
	this->_seqNumber++;
	close(_socket);
	_socket = 0;
	return 0;
	} else if (hdr.status == RESP_STOP_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) {
	logger_->log_trace("Flow Control Protocol stop flow controller");
	this->_controller->stop(true);
	this->_seqNumber++;
	close(_socket);
	_socket = 0;
	return 0;
	} else if (hdr.status == RESP_START_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) {
	logger_->log_trace("Flow Control Protocol start flow controller");
	this->_controller->start();
	this->_seqNumber++;
	close(_socket);
	_socket = 0;
	return 0;
	} else {
	logger_->log_trace("Flow Control Protocol Report fail");
	close(_socket);
	_socket = 0;
	return -1;
	}
	}

	} /* namespace minifi */
	} /* namespace nifi */
	} /* namespace apache */
	} /* namespace org */