You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by al...@apache.org on 2017/02/24 15:58:28 UTC
[5/5] nifi-minifi-cpp git commit: MINIFI-206: Implement Extensible
Comm Mechanism
MINIFI-206: Implement Extensible Comm Mechanism
Create initial communication mechanism that's
initially based on BSD sockets.
MINIFI-201: Change how hostname is resolved
and ensure that it is not null
MINIFI-199: Utilize zlib's crc32 implementation,
which is more portable and faster. we can explore
alternatives when that time arrives.
MINIFI-200: Put host "get" into initialization.
Any change wouldn't be picked up by a running process
most likely.
This closes #54.
Signed-off-by: Aldrin Piri <al...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/7d3e0441
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/7d3e0441
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/7d3e0441
Branch: refs/heads/master
Commit: 7d3e0441342825255781d1800c0178046f34254f
Parents: 939751c
Author: Marc Parisi <ph...@apache.org>
Authored: Tue Feb 14 08:38:05 2017 -0500
Committer: Aldrin Piri <al...@apache.org>
Committed: Fri Feb 24 10:56:29 2017 -0500
----------------------------------------------------------------------
CMakeLists.txt | 4 +-
libminifi/CMakeLists.txt | 17 +-
libminifi/include/AppendHostInfo.h | 4 +-
libminifi/include/Configure.h | 16 +-
libminifi/include/Connection.h | 4 +-
libminifi/include/Exception.h | 1 -
libminifi/include/ExecuteProcess.h | 4 +-
libminifi/include/FlowControlProtocol.h | 20 +-
libminifi/include/FlowController.h | 38 +-
libminifi/include/FlowFileRecord.h | 4 +-
libminifi/include/GetFile.h | 4 +-
libminifi/include/ListenSyslog.h | 6 +-
libminifi/include/LogAttribute.h | 4 +-
libminifi/include/Logger.h | 38 +-
libminifi/include/ProcessContext.h | 4 +-
libminifi/include/ProcessGroup.h | 58 +--
libminifi/include/ProcessSession.h | 6 +-
libminifi/include/Processor.h | 6 +-
libminifi/include/Property.h | 196 +++------
libminifi/include/Provenance.h | 39 +-
libminifi/include/PutFile.h | 6 +-
libminifi/include/RealTimeDataCollector.h | 4 +-
libminifi/include/RemoteProcessorGroupPort.h | 33 +-
libminifi/include/ResourceClaim.h | 4 +-
libminifi/include/SchedulingAgent.h | 10 +-
libminifi/include/Serializable.h | 294 --------------
libminifi/include/Site2SiteClientProtocol.h | 437 ++++++++++----------
libminifi/include/Site2SitePeer.h | 342 ++++++----------
libminifi/include/TailFile.h | 4 +-
libminifi/include/TimeUtil.h | 82 ----
libminifi/include/io/BaseStream.h | 150 +++++++
libminifi/include/io/CRCStream.h | 306 ++++++++++++++
libminifi/include/io/ClientSocket.h | 245 ++++++++++++
libminifi/include/io/DataStream.h | 120 ++++++
libminifi/include/io/EndianCheck.h | 47 +++
libminifi/include/io/Serializable.h | 186 +++++++++
libminifi/include/io/SocketFactory.h | 87 ++++
libminifi/include/io/TLSSocket.h | 185 +++++++++
libminifi/include/io/validation.h | 77 ++++
libminifi/include/utils/FailurePolicy.h | 45 +++
libminifi/include/utils/StringUtils.h | 125 ++++++
libminifi/include/utils/TimeUtil.h | 74 ++++
libminifi/src/AppendHostInfo.cpp | 14 +-
libminifi/src/Configure.cpp | 34 +-
libminifi/src/Connection.cpp | 12 +-
libminifi/src/ExecuteProcess.cpp | 21 +-
libminifi/src/FlowControlProtocol.cpp | 68 ++--
libminifi/src/FlowController.cpp | 190 +++------
libminifi/src/FlowFileRecord.cpp | 8 +-
libminifi/src/GenerateFlowFile.cpp | 3 +-
libminifi/src/GetFile.cpp | 29 +-
libminifi/src/ListenHTTP.cpp | 2 +-
libminifi/src/ListenSyslog.cpp | 17 +-
libminifi/src/LogAttribute.cpp | 15 +-
libminifi/src/Logger.cpp | 2 +-
libminifi/src/ProcessGroup.cpp | 259 ++++++------
libminifi/src/ProcessSession.cpp | 46 +--
libminifi/src/Processor.cpp | 40 +-
libminifi/src/Provenance.cpp | 26 +-
libminifi/src/PutFile.cpp | 23 +-
libminifi/src/RealTimeDataCollector.cpp | 77 ++--
libminifi/src/RemoteProcessorGroupPort.cpp | 55 ++-
libminifi/src/ResourceClaim.cpp | 6 +-
libminifi/src/SchedulingAgent.cpp | 6 +-
libminifi/src/Serializable.cpp | 365 -----------------
libminifi/src/Site2SiteClientProtocol.cpp | 228 ++++++-----
libminifi/src/Site2SitePeer.cpp | 461 +---------------------
libminifi/src/TailFile.cpp | 19 +-
libminifi/src/ThreadedSchedulingAgent.cpp | 26 +-
libminifi/src/io/BaseStream.cpp | 153 +++++++
libminifi/src/io/CRCStream.cpp | 24 ++
libminifi/src/io/ClientSocket.cpp | 425 ++++++++++++++++++++
libminifi/src/io/DataStream.cpp | 128 ++++++
libminifi/src/io/EndianCheck.cpp | 22 ++
libminifi/src/io/Serializable.cpp | 263 ++++++++++++
libminifi/src/io/SocketFactory.cpp | 24 ++
libminifi/src/io/TLSSocket.cpp | 236 +++++++++++
libminifi/test/unit/CRCTests.h | 78 ++++
libminifi/test/unit/PropertyTests.h | 106 +++++
libminifi/test/unit/SerializationTests.h | 5 +-
libminifi/test/unit/SocketTests.h | 190 +++++++++
libminifi/test/unit/Tests.cpp | 4 +
libminifi/test/unit/TimeUtilsTest.h | 30 ++
83 files changed, 4456 insertions(+), 2620 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 7ae1f8e..0f2c31b 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -33,8 +33,8 @@ IF(POLICY CMP0048)
ENDIF(POLICY CMP0048)
include(CheckCXXCompilerFlag)
-CHECK_CXX_COMPILER_FLAG("-std=c++11" COMPILER_SUPPORTS_CXX11)
-CHECK_CXX_COMPILER_FLAG("-std=c++0x" COMPILER_SUPPORTS_CXX0X)
+CHECK_CXX_COMPILER_FLAG("-std=c++11 " COMPILER_SUPPORTS_CXX11)
+CHECK_CXX_COMPILER_FLAG("-std=c++0x " COMPILER_SUPPORTS_CXX0X)
if(COMPILER_SUPPORTS_CXX11)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
elseif(COMPILER_SUPPORTS_CXX0X)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/libminifi/CMakeLists.txt b/libminifi/CMakeLists.txt
index 269492c..d80d63b 100644
--- a/libminifi/CMakeLists.txt
+++ b/libminifi/CMakeLists.txt
@@ -36,12 +36,23 @@ ENDIF(POLICY CMP0048)
set(CMAKE_CXX_STANDARD 11)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
+include(CheckCXXCompilerFlag)
+CHECK_CXX_COMPILER_FLAG("-std=c++11" COMPILER_SUPPORTS_CXX11)
+CHECK_CXX_COMPILER_FLAG("-std=c++0x" COMPILER_SUPPORTS_CXX0X)
+if(COMPILER_SUPPORTS_CXX11)
+ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -g")
+elseif(COMPILER_SUPPORTS_CXX0X)
+ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++0x")
+else()
+ message(STATUS "The compiler ${CMAKE_CXX_COMPILER} has no C++11 support. Please use a different C++ compiler.")
+endif()
+
include_directories(../include)
include_directories(../thirdparty/yaml-cpp-yaml-cpp-0.5.3/include)
include_directories(../thirdparty/civetweb-1.9.1/include)
include_directories(include)
-file(GLOB SOURCES "src/*.cpp")
+file(GLOB SOURCES "src/*.cpp" "src/io/*.cpp" "src/utils/*.cpp")
file(GLOB SPD_SOURCES "../include/spdlog/*")
# Workaround the limitations of having a
@@ -52,6 +63,10 @@ add_library(minifi STATIC ${SOURCES})
# Include Boost System
find_package(Boost COMPONENTS system REQUIRED)
target_link_libraries(minifi ${Boost_SYSTEM_LIBRARY})
+find_package(ZLIB REQUIRED)
+
+include_directories(${ZLIB_INCLUDE_DIRS})
+target_link_libraries (minifi ${ZLIB_LIBRARIES})
# Include LevelDB
find_package (Leveldb REQUIRED)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/AppendHostInfo.h
----------------------------------------------------------------------
diff --git a/libminifi/include/AppendHostInfo.h b/libminifi/include/AppendHostInfo.h
index 8cc1d41..5d62aae 100644
--- a/libminifi/include/AppendHostInfo.h
+++ b/libminifi/include/AppendHostInfo.h
@@ -35,7 +35,7 @@ public:
AppendHostInfo(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid)
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
//! Destructor
virtual ~AppendHostInfo()
@@ -61,7 +61,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
};
#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Configure.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Configure.h b/libminifi/include/Configure.h
index f4aae3c..a60a6d5 100644
--- a/libminifi/include/Configure.h
+++ b/libminifi/include/Configure.h
@@ -34,11 +34,11 @@ public:
//! Get the singleton logger instance
static Configure * getConfigure()
{
- if (!_configure)
+ if (!configure_)
{
- _configure = new Configure();
+ configure_ = new Configure();
}
- return _configure;
+ return configure_;
}
//! nifi.flow.configuration.file
static const char *nifi_flow_configuration_file;
@@ -79,10 +79,6 @@ public:
}
//! Get the config value
bool get(std::string key, std::string &value);
- // Trim String utils
- std::string trim(const std::string& s);
- std::string trimLeft(const std::string& s);
- std::string trimRight(const std::string& s);
//! Parse one line in configure file like key=value
void parseConfigureFileLine(char *buf);
//! Load Configure File
@@ -105,19 +101,19 @@ private:
//! Mutex for protection
std::mutex _mtx;
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Home location for this executable
std::string _minifiHome;
Configure()
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
virtual ~Configure()
{
}
- static Configure *_configure;
+ static Configure *configure_;
protected:
std::map<std::string,std::string> _properties;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Connection.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Connection.h b/libminifi/include/Connection.h
index dc6b94b..d372b77 100644
--- a/libminifi/include/Connection.h
+++ b/libminifi/include/Connection.h
@@ -43,7 +43,7 @@ public:
/*!
* Create a new processor
*/
- Connection(std::string name, uuid_t uuid = NULL, uuid_t srcUUID = NULL, uuid_t destUUID = NULL);
+ explicit Connection(std::string name, uuid_t uuid = NULL, uuid_t srcUUID = NULL, uuid_t destUUID = NULL);
//! Destructor
virtual ~Connection() {}
//! Set Connection Name
@@ -190,7 +190,7 @@ private:
//! Queue for the Flow File
std::queue<FlowFileRecord *> _queue;
//! Logger
- Logger *_logger;
+ Logger *logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
Connection(const Connection &parent);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Exception.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Exception.h b/libminifi/include/Exception.h
index d321454..1e02fa5 100644
--- a/libminifi/include/Exception.h
+++ b/libminifi/include/Exception.h
@@ -80,7 +80,6 @@ public:
return _whatStr.c_str();
}
-protected:
private:
//! Exception type
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/ExecuteProcess.h
----------------------------------------------------------------------
diff --git a/libminifi/include/ExecuteProcess.h b/libminifi/include/ExecuteProcess.h
index dce287a..83cf23c 100644
--- a/libminifi/include/ExecuteProcess.h
+++ b/libminifi/include/ExecuteProcess.h
@@ -46,7 +46,7 @@ public:
ExecuteProcess(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid)
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
_redirectErrorStream = false;
_batchDuration = 0;
_workingDir = ".";
@@ -94,7 +94,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Property
std::string _command;
std::string _commandArgument;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/FlowControlProtocol.h
----------------------------------------------------------------------
diff --git a/libminifi/include/FlowControlProtocol.h b/libminifi/include/FlowControlProtocol.h
index ebf3c8a..b3b0686 100644
--- a/libminifi/include/FlowControlProtocol.h
+++ b/libminifi/include/FlowControlProtocol.h
@@ -176,8 +176,8 @@ public:
*/
FlowControlProtocol(FlowController *controller) {
_controller = controller;
- _logger = Logger::getLogger();
- _configure = Configure::getConfigure();
+ logger_ = Logger::getLogger();
+ configure_ = Configure::getConfigure();
_socket = 0;
_serverName = "localhost";
_serverPort = DEFAULT_NIFI_SERVER_PORT;
@@ -190,22 +190,22 @@ public:
std::string value;
- if (_configure->get(Configure::nifi_server_name, value))
+ if (configure_->get(Configure::nifi_server_name, value))
{
_serverName = value;
- _logger->log_info("NiFi Server Name %s", _serverName.c_str());
+ logger_->log_info("NiFi Server Name %s", _serverName.c_str());
}
- if (_configure->get(Configure::nifi_server_port, value) && Property::StringToInt(value, _serverPort))
+ if (configure_->get(Configure::nifi_server_port, value) && Property::StringToInt(value, _serverPort))
{
- _logger->log_info("NiFi Server Port: [%d]", _serverPort);
+ logger_->log_info("NiFi Server Port: [%d]", _serverPort);
}
- if (_configure->get(Configure::nifi_server_report_interval, value))
+ 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);
+ logger_->log_info("NiFi server report interval: [%d] ms", _reportInterval);
}
}
else
@@ -304,9 +304,9 @@ private:
//! Mutex for protection
std::mutex _mtx;
//! Logger
- Logger *_logger = NULL;
+ Logger *logger_ = NULL;
//! Configure
- Configure *_configure = NULL;
+ Configure *configure_ = NULL;
//! NiFi server Name
std::string _serverName;
//! NiFi server port
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/FlowController.h
----------------------------------------------------------------------
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index 561cd1b..56dba03 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -85,30 +85,9 @@ public:
static const int DEFAULT_MAX_TIMER_DRIVEN_THREAD = 10;
static const int DEFAULT_MAX_EVENT_DRIVEN_THREAD = 5;
- //! passphase for the private file callback
- static int pemPassWordCb(char *buf, int size, int rwflag, void *userdata)
- {
- std::string passphrase;
-
- if (Configure::getConfigure()->get(Configure::nifi_security_client_pass_phrase, passphrase))
- {
- std::ifstream file(passphrase.c_str(), std::ifstream::in);
- if (!file.good())
- {
- memset(buf, 0, size);
- return 0;
- }
- memset(buf, 0, size);
- file.getline(buf, size);
- return (int) strlen(buf);
- }
- return 0;
- }
//! Destructor
virtual ~FlowController(){
- if (_ctx)
- SSL_CTX_free(_ctx);
}
//! Set FlowController Name
virtual void setName(std::string name) {
@@ -193,18 +172,9 @@ public:
_protocol->setSerialNumber(number);
}
-
- //! getSSLContext
- virtual SSL_CTX *getSSLContext()
- {
- return _ctx;
- }
protected:
- //! SSL context
- SSL_CTX *_ctx;
-
//! A global unique identifier
uuid_t _uuid;
//! FlowController Name
@@ -243,13 +213,13 @@ protected:
FlowController() :
_root(0), _maxTimerDrivenThreads(0), _maxEventDrivenThreads(0), _running(
false), _initialized(false), _provenanceRepo(0), _protocol(
- 0), _logger(Logger::getLogger()), _ctx(NULL){
+ 0), logger_(Logger::getLogger()){
}
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
};
@@ -311,8 +281,8 @@ private:
//! Mutex for protection
std::mutex _mtx;
//! Logger
- Logger *_logger;
- Configure *_configure;
+ Logger *logger_;
+ Configure *configure_;
//! Process Processor Node YAML
void parseProcessorNodeYaml(YAML::Node processorNode, ProcessGroup *parent);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/FlowFileRecord.h
----------------------------------------------------------------------
diff --git a/libminifi/include/FlowFileRecord.h b/libminifi/include/FlowFileRecord.h
index 9382996..6626572 100644
--- a/libminifi/include/FlowFileRecord.h
+++ b/libminifi/include/FlowFileRecord.h
@@ -31,7 +31,7 @@
#include <fstream>
#include <set>
-#include "TimeUtil.h"
+#include "utils/TimeUtil.h"
#include "Logger.h"
#include "ResourceClaim.h"
@@ -216,7 +216,7 @@ private:
//! Orginal connection queue that this flow file was dequeued from
Connection *_orginalConnection;
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Snapshot flow record for session rollback
bool _snapshot;
// Prevent default copy constructor and assignment operation
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/GetFile.h
----------------------------------------------------------------------
diff --git a/libminifi/include/GetFile.h b/libminifi/include/GetFile.h
index 4a8775f..0d74110 100644
--- a/libminifi/include/GetFile.h
+++ b/libminifi/include/GetFile.h
@@ -35,7 +35,7 @@ public:
GetFile(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid)
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
_directory = ".";
_recursive = true;
_keepSourceFile = false;
@@ -82,7 +82,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Queue for store directory list
std::queue<std::string> _dirList;
//! Get Listing size
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/ListenSyslog.h
----------------------------------------------------------------------
diff --git a/libminifi/include/ListenSyslog.h b/libminifi/include/ListenSyslog.h
index 81bc92c..309a3ad 100644
--- a/libminifi/include/ListenSyslog.h
+++ b/libminifi/include/ListenSyslog.h
@@ -53,7 +53,7 @@ public:
ListenSyslog(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid)
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
_eventQueueByteSize = 0;
_serverSocket = 0;
_recvBufSize = 65507;
@@ -87,7 +87,7 @@ public:
_clientSockets.clear();
if (_serverSocket > 0)
{
- _logger->log_info("ListenSysLog Server socket %d close", _serverSocket);
+ logger_->log_info("ListenSysLog Server socket %d close", _serverSocket);
close(_serverSocket);
_serverSocket = 0;
}
@@ -130,7 +130,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Run function for the thread
static void run(ListenSyslog *process);
//! Run Thread
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/LogAttribute.h
----------------------------------------------------------------------
diff --git a/libminifi/include/LogAttribute.h b/libminifi/include/LogAttribute.h
index 125ebf3..2478c7a 100644
--- a/libminifi/include/LogAttribute.h
+++ b/libminifi/include/LogAttribute.h
@@ -35,7 +35,7 @@ public:
LogAttribute(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid)
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
//! Destructor
virtual ~LogAttribute()
@@ -122,7 +122,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
};
#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Logger.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Logger.h b/libminifi/include/Logger.h
index 189ce6c..91d0e72 100644
--- a/libminifi/include/Logger.h
+++ b/libminifi/include/Logger.h
@@ -65,14 +65,14 @@ public:
//! Get the singleton logger instance
static Logger * getLogger() {
- if (!_logger)
- _logger = new Logger();
- return _logger;
+ if (!logger_)
+ logger_ = new Logger();
+ return logger_;
}
void setLogLevel(LOG_LEVEL_E level) {
- if (_spdlog == NULL)
+ if (spdlog_ == NULL)
return;
- _spdlog->set_level((spdlog::level::level_enum) level);
+ spdlog_->set_level((spdlog::level::level_enum) level);
}
void setLogLevel(const std::string &level,LOG_LEVEL_E defaultLevel = info )
@@ -112,10 +112,10 @@ public:
* @warning does not check @p log or @p format for null. Caller must ensure parameters and format string lengths match
*/
void log_error(const char *const format, ...) {
- if(_spdlog == NULL || !_spdlog->should_log(spdlog::level::level_enum::err))
+ if(spdlog_ == NULL || !spdlog_->should_log(spdlog::level::level_enum::err))
return;
FILL_BUFFER
- _spdlog->error(buffer);
+ spdlog_->error(buffer);
}
/**
* @brief Log warn message
@@ -123,10 +123,10 @@ public:
* @warning does not check @p log or @p format for null. Caller must ensure parameters and format string lengths match
*/
void log_warn(const char *const format, ...) {
- if(_spdlog == NULL || !_spdlog->should_log(spdlog::level::level_enum::warn))
+ if(spdlog_ == NULL || !spdlog_->should_log(spdlog::level::level_enum::warn))
return;
FILL_BUFFER
- _spdlog->warn(buffer);
+ spdlog_->warn(buffer);
}
/**
* @brief Log info message
@@ -134,10 +134,10 @@ public:
* @warning does not check @p log or @p format for null. Caller must ensure parameters and format string lengths match
*/
void log_info(const char *const format, ...) {
- if(_spdlog == NULL || !_spdlog->should_log(spdlog::level::level_enum::info))
+ if(spdlog_ == NULL || !spdlog_->should_log(spdlog::level::level_enum::info))
return;
FILL_BUFFER
- _spdlog->info(buffer);
+ spdlog_->info(buffer);
}
/**
* @brief Log debug message
@@ -145,10 +145,10 @@ public:
* @warning does not check @p log or @p format for null. Caller must ensure parameters and format string lengths match
*/
void log_debug(const char *const format, ...) {
- if(_spdlog == NULL || !_spdlog->should_log(spdlog::level::level_enum::debug))
+ if(spdlog_ == NULL || !spdlog_->should_log(spdlog::level::level_enum::debug))
return;
FILL_BUFFER
- _spdlog->debug(buffer);
+ spdlog_->debug(buffer);
}
/**
* @brief Log trace message
@@ -156,10 +156,10 @@ public:
* @warning does not check @p log or @p format for null. Caller must ensure parameters and format string lengths match
*/
void log_trace(const char *const format, ...) {
- if(_spdlog == NULL || !_spdlog->should_log(spdlog::level::level_enum::trace))
+ if(spdlog_ == NULL || !spdlog_->should_log(spdlog::level::level_enum::trace))
return;
FILL_BUFFER
- _spdlog->trace(buffer);
+ spdlog_->trace(buffer);
}
protected:
@@ -174,14 +174,14 @@ private:
* Create a logger
* */
Logger(const std::string logger_name = LOG_NAME, const std::string filename = LOG_FILE_NAME, size_t max_file_size = DEFAULT_LOG_FILE_SIZE, size_t max_files = DEFAULT_LOG_FILE_NUMBER, bool force_flush = true) {
- _spdlog = rotating_logger_mt(logger_name, filename, max_file_size, max_files);
- _spdlog->set_level((spdlog::level::level_enum) debug);
+ spdlog_ = rotating_logger_mt(logger_name, filename, max_file_size, max_files);
+ spdlog_->set_level((spdlog::level::level_enum) debug);
}
//! spdlog
- std::shared_ptr<logger> _spdlog;
+ std::shared_ptr<logger> spdlog_;
//! Singleton logger instance
- static Logger *_logger;
+ static Logger *logger_;
};
#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/ProcessContext.h
----------------------------------------------------------------------
diff --git a/libminifi/include/ProcessContext.h b/libminifi/include/ProcessContext.h
index 5c3288e..469c87e 100644
--- a/libminifi/include/ProcessContext.h
+++ b/libminifi/include/ProcessContext.h
@@ -40,7 +40,7 @@ public:
* Create a new process context associated with the processor/controller service/state manager
*/
ProcessContext(Processor *processor = NULL) : _processor(processor) {
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
//! Destructor
virtual ~ProcessContext() {}
@@ -106,7 +106,7 @@ private:
//! Processor
Processor *_processor;
//! Logger
- Logger *_logger;
+ Logger *logger_;
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/ProcessGroup.h
----------------------------------------------------------------------
diff --git a/libminifi/include/ProcessGroup.h b/libminifi/include/ProcessGroup.h
index 304cfe6..28661ae 100644
--- a/libminifi/include/ProcessGroup.h
+++ b/libminifi/include/ProcessGroup.h
@@ -56,56 +56,56 @@ public:
virtual ~ProcessGroup();
//! Set Processor Name
void setName(std::string name) {
- _name = name;
+ name_ = name;
}
//! Get Process Name
std::string getName(void) {
- return (_name);
+ return (name_);
}
//! Set URL
void setURL(std::string url) {
- _url = url;
+ url_ = url;
}
//! Get URL
std::string getURL(void) {
- return (_url);
+ return (url_);
}
//! SetTransmitting
void setTransmitting(bool val)
{
- _transmitting = val;
+ transmitting_ = val;
}
//! Get Transmitting
bool getTransmitting()
{
- return _transmitting;
+ return transmitting_;
}
//! setTimeOut
void setTimeOut(uint64_t time)
{
- _timeOut = time;
+ timeOut_ = time;
}
uint64_t getTimeOut()
{
- return _timeOut;
+ return timeOut_;
}
//! Set Processor yield period in MilliSecond
void setYieldPeriodMsec(uint64_t period) {
- _yieldPeriodMsec = period;
+ yield_period_msec_ = period;
}
//! Get Processor yield period in MilliSecond
uint64_t getYieldPeriodMsec(void) {
- return(_yieldPeriodMsec);
+ return(yield_period_msec_);
}
//! Set UUID
void setUUID(uuid_t uuid) {
- uuid_copy(_uuid, uuid);
+ uuid_copy(uuid_, uuid);
}
//! Get UUID
bool getUUID(uuid_t uuid) {
if (uuid)
{
- uuid_copy(uuid, _uuid);
+ uuid_copy(uuid, uuid_);
return true;
}
else
@@ -121,13 +121,13 @@ public:
bool isRootProcessGroup();
//! set parent process group
void setParent(ProcessGroup *parent) {
- std::lock_guard<std::mutex> lock(_mtx);
- _parentProcessGroup = parent;
+ std::lock_guard<std::mutex> lock(mtx_);
+ parent_process_group_ = parent;
}
//! get parent process group
ProcessGroup *getParent(void) {
- std::lock_guard<std::mutex> lock(_mtx);
- return _parentProcessGroup;
+ std::lock_guard<std::mutex> lock(mtx_);
+ return parent_process_group_;
}
//! Add processor
void addProcessor(Processor *processor);
@@ -150,32 +150,32 @@ public:
protected:
//! A global unique identifier
- uuid_t _uuid;
+ uuid_t uuid_;
//! Processor Group Name
- std::string _name;
+ std::string name_;
//! Process Group Type
- ProcessGroupType _type;
+ ProcessGroupType type_;
//! Processors (ProcessNode) inside this process group which include Input/Output Port, Remote Process Group input/Output port
- std::set<Processor *> _processors;
- std::set<ProcessGroup *> _childProcessGroups;
+ std::set<Processor *> processors_;
+ std::set<ProcessGroup *> child_process_groups_;
//! Connections between the processor inside the group;
- std::set<Connection *> _connections;
+ std::set<Connection *> connections_;
//! Parent Process Group
- ProcessGroup* _parentProcessGroup;
+ ProcessGroup* parent_process_group_;
//! Yield Period in Milliseconds
- std::atomic<uint64_t> _yieldPeriodMsec;
- std::atomic<uint64_t> _timeOut;
+ std::atomic<uint64_t> yield_period_msec_;
+ std::atomic<uint64_t> timeOut_;
//! URL
- std::string _url;
+ std::string url_;
//! Transmitting
- std::atomic<bool> _transmitting;
+ std::atomic<bool> transmitting_;
private:
//! Mutex for protection
- std::mutex _mtx;
+ std::mutex mtx_;
//! Logger
- Logger *_logger;
+ Logger *logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
ProcessGroup(const ProcessGroup &parent);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/ProcessSession.h
----------------------------------------------------------------------
diff --git a/libminifi/include/ProcessSession.h b/libminifi/include/ProcessSession.h
index 44032c3..7146553 100644
--- a/libminifi/include/ProcessSession.h
+++ b/libminifi/include/ProcessSession.h
@@ -45,8 +45,8 @@ public:
* Create a new process session
*/
ProcessSession(ProcessContext *processContext = NULL) : _processContext(processContext) {
- _logger = Logger::getLogger();
- _logger->log_trace("ProcessSession created for %s", _processContext->getProcessor()->getName().c_str());
+ logger_ = Logger::getLogger();
+ logger_->log_trace("ProcessSession created for %s", _processContext->getProcessor()->getName().c_str());
_provenanceReport = new ProvenanceReporter(_processContext->getProcessor()->getUUIDStr(),
_processContext->getProcessor()->getName());
}
@@ -121,7 +121,7 @@ private:
//! ProcessContext
ProcessContext *_processContext;
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Provenance Report
ProvenanceReporter *_provenanceReport;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Processor.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Processor.h b/libminifi/include/Processor.h
index 95f47c1..b675b7c 100644
--- a/libminifi/include/Processor.h
+++ b/libminifi/include/Processor.h
@@ -33,7 +33,7 @@
#include <chrono>
#include <functional>
-#include "TimeUtil.h"
+#include "utils/TimeUtil.h"
#include "Property.h"
#include "Relationship.h"
#include "Connection.h"
@@ -119,7 +119,9 @@ public:
return true;
}
else
+ {
return false;
+ }
}
//! Set the supported processor properties while the process is not running
bool setSupportedProperties(std::set<Property> properties);
@@ -352,7 +354,7 @@ private:
//! Check all incoming connections for work
bool isWorkAvailable();
//! Logger
- Logger *_logger;
+ Logger *logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
Processor(const Processor &parent);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Property.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Property.h b/libminifi/include/Property.h
index a724394..93f9c0e 100644
--- a/libminifi/include/Property.h
+++ b/libminifi/include/Property.h
@@ -20,24 +20,22 @@
#ifndef __PROPERTY_H__
#define __PROPERTY_H__
+#include <algorithm>
+#include <sstream>
#include <string>
#include <vector>
#include <queue>
#include <map>
#include <mutex>
#include <atomic>
+#include <functional>
#include <set>
#include <stdlib.h>
#include <math.h>
//! Time Unit
enum TimeUnit {
- DAY,
- HOUR,
- MINUTE,
- SECOND,
- MILLISECOND,
- NANOSECOND
+ DAY, HOUR, MINUTE, SECOND, MILLISECOND, NANOSECOND
};
//! Property Class
@@ -48,12 +46,15 @@ public:
/*!
* Create a new property
*/
- Property(const std::string name, const std::string description, const std::string value)
- : _name(name), _description(description), _value(value) {
+ Property(const std::string name, const std::string description,
+ const std::string value) :
+ _name(name), _description(description), _value(value) {
+ }
+ Property() {
}
- Property() {}
//! Destructor
- virtual ~Property() {}
+ virtual ~Property() {
+ }
//! Get Name for the property
std::string getName() {
return _name;
@@ -71,84 +72,60 @@ public:
_value = value;
}
//! Compare
- bool operator < (const Property & right) const {
+ bool operator <(const Property & right) const {
return _name < right._name;
}
//! Convert TimeUnit to MilliSecond
- static bool ConvertTimeUnitToMS(int64_t input, TimeUnit unit, int64_t &out)
- {
- if (unit == MILLISECOND)
- {
+ static bool ConvertTimeUnitToMS(int64_t input, TimeUnit unit,
+ int64_t &out) {
+ if (unit == MILLISECOND) {
out = input;
return true;
- }
- else if (unit == SECOND)
- {
+ } else if (unit == SECOND) {
out = input * 1000;
return true;
- }
- else if (unit == MINUTE)
- {
+ } else if (unit == MINUTE) {
out = input * 60 * 1000;
return true;
- }
- else if (unit == HOUR)
- {
+ } else if (unit == HOUR) {
out = input * 60 * 60 * 1000;
return true;
- }
- else if (unit == DAY)
- {
+ } else if (unit == DAY) {
out = 24 * 60 * 60 * 1000;
return true;
- }
- else if (unit == NANOSECOND)
- {
- out = input/1000/1000;
+ } else if (unit == NANOSECOND) {
+ out = input / 1000 / 1000;
return true;
- }
- else
- {
+ } else {
return false;
}
}
//! Convert TimeUnit to NanoSecond
- static bool ConvertTimeUnitToNS(int64_t input, TimeUnit unit, int64_t &out)
- {
- if (unit == MILLISECOND)
- {
+ static bool ConvertTimeUnitToNS(int64_t input, TimeUnit unit,
+ int64_t &out) {
+ if (unit == MILLISECOND) {
out = input * 1000 * 1000;
return true;
- }
- else if (unit == SECOND)
- {
+ } else if (unit == SECOND) {
out = input * 1000 * 1000 * 1000;
return true;
- }
- else if (unit == MINUTE)
- {
+ } else if (unit == MINUTE) {
out = input * 60 * 1000 * 1000 * 1000;
return true;
- }
- else if (unit == HOUR)
- {
+ } else if (unit == HOUR) {
out = input * 60 * 60 * 1000 * 1000 * 1000;
return true;
- }
- else if (unit == NANOSECOND)
- {
+ } else if (unit == NANOSECOND) {
out = input;
return true;
- }
- else
- {
+ } else {
return false;
}
}
//! Convert String
- static bool StringToTime(std::string input, int64_t &output, TimeUnit &timeunit)
- {
+ static bool StringToTime(std::string input, int64_t &output,
+ TimeUnit &timeunit) {
if (input.size() == 0) {
return false;
}
@@ -157,62 +134,52 @@ public:
char *pEnd;
long int ival = strtol(cvalue, &pEnd, 0);
- if (pEnd[0] == '\0')
- {
+ if (pEnd[0] == '\0') {
return false;
}
- while (*pEnd == ' ')
- {
+ while (*pEnd == ' ') {
// Skip the space
pEnd++;
}
std::string unit(pEnd);
- if (unit == "sec" || unit == "s" || unit == "second" || unit == "seconds" || unit == "secs")
- {
+ if (unit == "sec" || unit == "s" || unit == "second"
+ || unit == "seconds" || unit == "secs") {
timeunit = SECOND;
output = ival;
return true;
- }
- else if (unit == "min" || unit == "m" || unit == "mins" || unit == "minute" || unit == "minutes")
- {
+ } else if (unit == "min" || unit == "m" || unit == "mins"
+ || unit == "minute" || unit == "minutes") {
timeunit = MINUTE;
output = ival;
return true;
- }
- else if (unit == "ns" || unit == "nano" || unit == "nanos" || unit == "nanoseconds")
- {
+ } else if (unit == "ns" || unit == "nano" || unit == "nanos"
+ || unit == "nanoseconds") {
timeunit = NANOSECOND;
output = ival;
return true;
- }
- else if (unit == "ms" || unit == "milli" || unit == "millis" || unit == "milliseconds")
- {
+ } else if (unit == "ms" || unit == "milli" || unit == "millis"
+ || unit == "milliseconds") {
timeunit = MILLISECOND;
output = ival;
return true;
- }
- else if (unit == "h" || unit == "hr" || unit == "hour" || unit == "hrs" || unit == "hours")
- {
+ } else if (unit == "h" || unit == "hr" || unit == "hour"
+ || unit == "hrs" || unit == "hours") {
timeunit = HOUR;
output = ival;
return true;
- }
- else if (unit == "d" || unit == "day" || unit == "days")
- {
+ } else if (unit == "d" || unit == "day" || unit == "days") {
timeunit = DAY;
output = ival;
return true;
- }
- else
+ } else
return false;
}
//! Convert String to Integer
- static bool StringToInt(std::string input, int64_t &output)
- {
+ static bool StringToInt(std::string input, int64_t &output) {
if (input.size() == 0) {
return false;
}
@@ -221,26 +188,23 @@ public:
char *pEnd;
long int ival = strtol(cvalue, &pEnd, 0);
- if (pEnd[0] == '\0')
- {
+ if (pEnd[0] == '\0') {
output = ival;
return true;
}
- while (*pEnd == ' ')
- {
+ while (*pEnd == ' ') {
// Skip the space
pEnd++;
}
char end0 = toupper(pEnd[0]);
- if ( (end0 == 'K') || (end0 == 'M') || (end0 == 'G') || (end0 == 'T') || (end0 == 'P') )
- {
- if (pEnd[1] == '\0')
- {
+ if ((end0 == 'K') || (end0 == 'M') || (end0 == 'G') || (end0 == 'T')
+ || (end0 == 'P')) {
+ if (pEnd[1] == '\0') {
unsigned long int multiplier = 1000;
- if ( (end0 != 'K') ) {
+ if ((end0 != 'K')) {
multiplier *= 1000;
if (end0 != 'M') {
multiplier *= 1000;
@@ -255,11 +219,12 @@ public:
output = ival * multiplier;
return true;
- } else if ((pEnd[1] == 'b' || pEnd[1] == 'B') && (pEnd[2] == '\0')) {
+ } else if ((pEnd[1] == 'b' || pEnd[1] == 'B')
+ && (pEnd[2] == '\0')) {
unsigned long int multiplier = 1024;
- if ( (end0 != 'K') ) {
+ if ((end0 != 'K')) {
multiplier *= 1024;
if (end0 != 'M') {
multiplier *= 1024;
@@ -278,56 +243,7 @@ public:
return false;
}
- //! Convert String to Float
- static bool StringToFloat(std::string input, float &output)
- {
- const char *cvalue = input.c_str();
- char *pEnd;
- float val = strtof(cvalue, &pEnd);
-
- if (pEnd[0] == '\0')
- {
- output = val;
- return true;
- }
- else
- return false;
- }
- //! Convert String to Bool
- static bool StringToBool(std::string input, bool &output)
- {
- if (input == "true" || input == "True" || input == "TRUE")
- {
- output = true;
- return true;
- }
- if (input == "false" || input == "False" || input == "FALSE")
- {
- output = false;
- return true;
- }
- return false;
- }
- // Trim String utils
- static std::string trim(const std::string& s)
- {
- return trimRight(trimLeft(s));
- }
-
- static std::string trimLeft(const std::string& s)
- {
- const char *WHITESPACE = " \n\r\t";
- size_t startpos = s.find_first_not_of(WHITESPACE);
- return (startpos == std::string::npos) ? "" : s.substr(startpos);
- }
-
- static std::string trimRight(const std::string& s)
- {
- const char *WHITESPACE = " \n\r\t";
- size_t endpos = s.find_last_not_of(WHITESPACE);
- return (endpos == std::string::npos) ? "" : s.substr(0, endpos+1);
- }
protected:
//! Name
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Provenance.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Provenance.h b/libminifi/include/Provenance.h
index 822b3cf..10945c3 100644
--- a/libminifi/include/Provenance.h
+++ b/libminifi/include/Provenance.h
@@ -42,8 +42,8 @@
#include "Logger.h"
#include "Property.h"
#include "ResourceClaim.h"
-#include "TimeUtil.h"
-#include "Serializable.h"
+#include "io/Serializable.h"
+#include "utils/TimeUtil.h"
// Provenance Event Record Serialization Seg Size
#define PROVENANCE_EVENT_RECORD_SEG_SIZE 2048
@@ -173,12 +173,12 @@ public:
uuid_generate(_eventId);
uuid_unparse_lower(_eventId, eventIdStr);
_eventIdStr = eventIdStr;
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
ProvenanceEventRecord() {
_eventTime = getTimeMillis();
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
//! Destructor
@@ -451,7 +451,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
@@ -470,7 +470,7 @@ public:
* Create a new provenance reporter associated with the process session
*/
ProvenanceReporter(std::string componentId, std::string componentType) {
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
_componentId = componentId;
_componentType = componentType;
}
@@ -555,7 +555,7 @@ private:
//! Incoming connection Iterator
std::set<ProvenanceEventRecord *> _events;
//! Logger
- Logger *_logger;
+ Logger *logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
@@ -577,8 +577,8 @@ public:
* Create a new provenance repository
*/
ProvenanceRepository() {
- _logger = Logger::getLogger();
- _configure = Configure::getConfigure();
+ logger_ = Logger::getLogger();
+ configure_ = Configure::getConfigure();
_directory = PROVENANCE_DIRECTORY;
_maxPartitionMillis = MAX_PROVENANCE_ENTRY_LIFE_TIME;
_purgePeriod = PROVENANCE_PURGE_PERIOD;
@@ -601,17 +601,17 @@ public:
virtual bool initialize()
{
std::string value;
- if (_configure->get(Configure::nifi_provenance_repository_directory_default, value))
+ if (configure_->get(Configure::nifi_provenance_repository_directory_default, value))
{
_directory = value;
}
- _logger->log_info("NiFi Provenance Repository Directory %s", _directory.c_str());
- if (_configure->get(Configure::nifi_provenance_repository_max_storage_size, value))
+ logger_->log_info("NiFi Provenance Repository Directory %s", _directory.c_str());
+ if (configure_->get(Configure::nifi_provenance_repository_max_storage_size, value))
{
Property::StringToInt(value, _maxPartitionBytes);
}
- _logger->log_info("NiFi Provenance Max Partition Bytes %d", _maxPartitionBytes);
- if (_configure->get(Configure::nifi_provenance_repository_max_storage_time, value))
+ logger_->log_info("NiFi Provenance Max Partition Bytes %d", _maxPartitionBytes);
+ if (configure_->get(Configure::nifi_provenance_repository_max_storage_time, value))
{
TimeUnit unit;
if (Property::StringToTime(value, _maxPartitionMillis, unit) &&
@@ -619,17 +619,17 @@ public:
{
}
}
- _logger->log_info("NiFi Provenance Max Storage Time: [%d] ms", _maxPartitionMillis);
+ logger_->log_info("NiFi Provenance Max Storage Time: [%d] ms", _maxPartitionMillis);
leveldb::Options options;
options.create_if_missing = true;
leveldb::Status status = leveldb::DB::Open(options, _directory.c_str(), &_db);
if (status.ok())
{
- _logger->log_info("NiFi Provenance Repository database open %s success", _directory.c_str());
+ logger_->log_info("NiFi Provenance Repository database open %s success", _directory.c_str());
}
else
{
- _logger->log_error("NiFi Provenance Repository database open %s fail", _directory.c_str());
+ logger_->log_error("NiFi Provenance Repository database open %s fail", _directory.c_str());
return false;
}
@@ -640,6 +640,7 @@ public:
//! Put
virtual bool Put(std::string key, uint8_t *buf, int bufLen)
{
+
// persistent to the DB
leveldb::Slice value((const char *) buf, bufLen);
leveldb::Status status;
@@ -709,10 +710,10 @@ private:
//! repository directory
std::string _directory;
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Configure
//! max db entry life time
- Configure *_configure;
+ Configure *configure_;
int64_t _maxPartitionMillis;
//! max db size
int64_t _maxPartitionBytes;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/PutFile.h
----------------------------------------------------------------------
diff --git a/libminifi/include/PutFile.h b/libminifi/include/PutFile.h
index 9f1375d..bf48f79 100644
--- a/libminifi/include/PutFile.h
+++ b/libminifi/include/PutFile.h
@@ -40,7 +40,7 @@ public:
PutFile(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid)
{
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
}
//! Destructor
virtual ~PutFile()
@@ -69,7 +69,7 @@ public:
bool commit();
private:
- Logger *_logger;
+ Logger *logger_;
std::ofstream _tmpFileOs;
bool _writeSucceeded = false;
std::string _tmpFile;
@@ -80,7 +80,7 @@ protected:
private:
//! Logger
- Logger *_logger;
+ Logger *logger_;
bool putFile(ProcessSession *session, FlowFileRecord *flowFile, const std::string &tmpFile, const std::string &destFile);
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/RealTimeDataCollector.h
----------------------------------------------------------------------
diff --git a/libminifi/include/RealTimeDataCollector.h b/libminifi/include/RealTimeDataCollector.h
index 760b566..6458f6d 100644
--- a/libminifi/include/RealTimeDataCollector.h
+++ b/libminifi/include/RealTimeDataCollector.h
@@ -47,7 +47,7 @@ public:
{
_realTimeSocket = 0;
_batchSocket = 0;
- _logger = Logger::getLogger();
+ logger_ = Logger::getLogger();
_firstInvoking = false;
_realTimeAccumulated = 0;
_batchAcccumulated = 0;
@@ -113,7 +113,7 @@ private:
int _realTimeSocket;
int _batchSocket;
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Mutex for protection
std::mutex _mtx;
//! Queued data size
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/RemoteProcessorGroupPort.h
----------------------------------------------------------------------
diff --git a/libminifi/include/RemoteProcessorGroupPort.h b/libminifi/include/RemoteProcessorGroupPort.h
index cd99e50..73e8ac2 100644
--- a/libminifi/include/RemoteProcessorGroupPort.h
+++ b/libminifi/include/RemoteProcessorGroupPort.h
@@ -20,6 +20,7 @@
#ifndef __REMOTE_PROCESSOR_GROUP_PORT_H__
#define __REMOTE_PROCESSOR_GROUP_PORT_H__
+#include <memory>
#include "FlowFileRecord.h"
#include "Processor.h"
#include "ProcessSession.h"
@@ -34,18 +35,16 @@ public:
* Create a new processor
*/
RemoteProcessorGroupPort(std::string name, uuid_t uuid = NULL)
- : Processor(name, uuid)
+ : Processor(name, uuid), direction_(SEND),transmitting_(false), peer_()
{
- _logger = Logger::getLogger();
- _peer = new Site2SitePeer("", 9999);
- _protocol = new Site2SiteClientProtocol(_peer);
- _protocol->setPortId(uuid);
+ logger_ = Logger::getLogger();
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(new Site2SiteClientProtocol(0));
+ protocol_->setPortId(uuid);
}
//! Destructor
virtual ~RemoteProcessorGroupPort()
{
- delete _protocol;
- delete _peer;
+
}
//! Processor Name
static const std::string ProcessorName;
@@ -62,34 +61,34 @@ public:
//! Set Direction
void setDirection(TransferDirection direction)
{
- _direction = direction;
- if (_direction == RECEIVE)
+ direction_ = direction;
+ if (direction_ == RECEIVE)
this->setTriggerWhenEmpty(true);
}
//! Set Timeout
void setTimeOut(uint64_t timeout)
{
- _protocol->setTimeOut(timeout);
+ protocol_->setTimeOut(timeout);
}
//! SetTransmitting
void setTransmitting(bool val)
{
- _transmitting = val;
+ transmitting_ = val;
}
protected:
private:
//! Logger
- Logger *_logger;
- //! Peer Connection
- Site2SitePeer *_peer;
+ Logger *logger_;
+
+ Site2SitePeer peer_;
//! Peer Protocol
- Site2SiteClientProtocol *_protocol;
+ std::unique_ptr<Site2SiteClientProtocol> protocol_;
//! Transaction Direction
- TransferDirection _direction;
+ TransferDirection direction_;
//! Transmitting
- bool _transmitting;
+ bool transmitting_;
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/ResourceClaim.h
----------------------------------------------------------------------
diff --git a/libminifi/include/ResourceClaim.h b/libminifi/include/ResourceClaim.h
index b4085f2..dcbb558 100644
--- a/libminifi/include/ResourceClaim.h
+++ b/libminifi/include/ResourceClaim.h
@@ -81,9 +81,9 @@ protected:
private:
//! Configure
- Configure *_configure;
+ Configure *configure_;
//! Logger
- Logger *_logger;
+ Logger *logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
ResourceClaim(const ResourceClaim &parent);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/SchedulingAgent.h
----------------------------------------------------------------------
diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h
index 1f94585..c814fa3 100644
--- a/libminifi/include/SchedulingAgent.h
+++ b/libminifi/include/SchedulingAgent.h
@@ -27,7 +27,7 @@
#include <atomic>
#include <algorithm>
#include <thread>
-#include "TimeUtil.h"
+#include "utils/TimeUtil.h"
#include "Logger.h"
#include "Configure.h"
#include "FlowFileRecord.h"
@@ -43,8 +43,8 @@ public:
* Create a new processor
*/
SchedulingAgent() {
- _configure = Configure::getConfigure();
- _logger = Logger::getLogger();
+ configure_ = Configure::getConfigure();
+ logger_ = Logger::getLogger();
_running = false;
}
//! Destructor
@@ -75,9 +75,9 @@ public:
protected:
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Configure
- Configure *_configure;
+ Configure *configure_;
//! Mutex for protection
std::mutex _mtx;
//! Whether it is running
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Serializable.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Serializable.h b/libminifi/include/Serializable.h
deleted file mode 100644
index c32dee0..0000000
--- a/libminifi/include/Serializable.h
+++ /dev/null
@@ -1,294 +0,0 @@
-/**
- *
- * 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 __SERIALIZABLE_H__
-#define __SERIALIZABLE_H__
-
-#include <cstdint>
-#include <vector>
-#include <string>
-
-
-/**
- * Mechanism to determine endianness of host.
- * Accounts for only BIG/LITTLE/BIENDIAN
- **/
-class EndiannessCheck
-{
-public:
- static bool IS_LITTLE;
-private:
-
- static bool is_little_endian() {
- /* do whatever is needed at static init time */
- unsigned int x = 1;
- char *c = (char*) &x;
- IS_LITTLE=*c==1;
- return IS_LITTLE;
- }
-};
-
-
-/**
- * DataStream defines the mechanism through which
- * binary data will be written to a sink
- */
-class DataStream
-{
-public:
-
- DataStream() : readBuffer(0)
- {
-
- }
-
- /**
- * Constructor
- **/
- DataStream(const uint8_t *buf, const uint32_t buflen) : DataStream()
- {
- writeData((uint8_t*)buf,buflen);
-
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector<uint8_t> &buf, int buflen);
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(uint8_t *buf, int buflen);
-
- /**
- * writes valiue to buffer
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size);
-
-
- /**
- * Reads a system word
- * @param value value to write
- */
- inline int readLongLong(uint64_t &value, bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- inline int readLong(uint32_t &value, bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
-
- /**
- * Reads a system short
- * @param value value to write
- */
- inline int readShort(uint16_t &value,bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
-
- /**
- * Returns the underlying buffer
- * @return vector's array
- **/
- const uint8_t *getBuffer() const
- {
- return &buffer[0];
- }
-
- /**
- * Retrieve size of data stream
- * @return size of data stream
- **/
- const uint32_t getSize() const
- {
- return buffer.size();
- }
-
-private:
- // All serialization related method and internal buf
- std::vector<uint8_t> buffer;
- uint32_t readBuffer;
-};
-
-/**
- * Serializable instances provide base functionality to
- * write certain objects/primitives to a data stream.
- *
- */
-class Serializable {
-
-public:
-
- /**
- * Inline function to write T to stream
- **/
- template<typename T>
- inline int writeData(const T &t,DataStream *stream);
-
- /**
- * Inline function to write T to to_vec
- **/
- template<typename T>
- inline int writeData(const T &t, uint8_t *to_vec);
-
- /**
- * Inline function to write T to to_vec
- **/
- template<typename T>
- inline int writeData(const T &t, std::vector<uint8_t> &to_vec);
-
-
- /**
- * write byte to stream
- * @return resulting write size
- **/
- int write(uint8_t value,DataStream *stream);
-
- /**
- * write byte to stream
- * @return resulting write size
- **/
- int write(char value,DataStream *stream);
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint32_t base_value,DataStream *stream, bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
- /**
- * write 2 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value,DataStream *stream, bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
- /**
- * write valueto stream
- * @param value non encoded value
- * @param len length of value
- * @param strema output stream
- * @return resulting write size
- **/
- int write(uint8_t *value, int len,DataStream *stream);
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value,DataStream *stream, bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
- /**
- * write bool to stream
- * @param value non encoded value
- * @return resulting write size
- **/
- int write(bool value);
-
- /**
- * write UTF string to stream
- * @param str string to write
- * @return resulting write size
- **/
- int writeUTF(std::string str,DataStream *stream, bool widen = false);
-
- /**
- * reads a byte from the stream
- * @param value reference in which will set the result
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int read(uint8_t &value,DataStream *stream);
-
- /**
- * reads two bytes from the stream
- * @param value reference in which will set the result
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int read(uint16_t &base_value,DataStream *stream, bool is_little_endian =
- EndiannessCheck::IS_LITTLE);
-
- /**
- * reads a byte from the stream
- * @param value reference in which will set the result
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int read(char &value,DataStream *stream);
-
- /**
- * reads a byte array from the stream
- * @param value reference in which will set the result
- * @param len length to read
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int read(uint8_t *value, int len,DataStream *stream);
-
- /**
- * reads four bytes from the stream
- * @param value reference in which will set the result
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int read(uint32_t &value,DataStream *stream,
- bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * reads eight byte from the stream
- * @param value reference in which will set the result
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int read(uint64_t &value,DataStream *stream,
- bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * read UTF from stream
- * @param str reference string
- * @param stream stream from which we will read
- * @return resulting read size
- **/
- int readUTF(std::string &str,DataStream *stream, bool widen = false);
-
-protected:
-
-};
-
-#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7d3e0441/libminifi/include/Site2SiteClientProtocol.h
----------------------------------------------------------------------
diff --git a/libminifi/include/Site2SiteClientProtocol.h b/libminifi/include/Site2SiteClientProtocol.h
index 1163436..fc1c411 100644
--- a/libminifi/include/Site2SiteClientProtocol.h
+++ b/libminifi/include/Site2SiteClientProtocol.h
@@ -41,6 +41,7 @@
#include "FlowFileRecord.h"
#include "ProcessContext.h"
#include "ProcessSession.h"
+#include "io/CRCStream.h"
//! Resource Negotiated Status Code
#define RESOURCE_OK 20
@@ -64,7 +65,6 @@ typedef enum {
RECEIVE
} TransferDirection;
-
//! Peer State
typedef enum {
/**
@@ -116,55 +116,50 @@ typedef enum {
//! Request Type
typedef enum {
- NEGOTIATE_FLOWFILE_CODEC = 0,
- REQUEST_PEER_LIST,
- SEND_FLOWFILES,
- RECEIVE_FLOWFILES,
- SHUTDOWN,
+ NEGOTIATE_FLOWFILE_CODEC = 0,
+ REQUEST_PEER_LIST,
+ SEND_FLOWFILES,
+ RECEIVE_FLOWFILES,
+ SHUTDOWN,
MAX_REQUEST_TYPE
} RequestType;
//! Request Type Str
-static const char *RequestTypeStr[MAX_REQUEST_TYPE] =
-{
- "NEGOTIATE_FLOWFILE_CODEC",
- "REQUEST_PEER_LIST",
- "SEND_FLOWFILES",
- "RECEIVE_FLOWFILES",
- "SHUTDOWN"
-};
+static const char *RequestTypeStr[MAX_REQUEST_TYPE] = {
+ "NEGOTIATE_FLOWFILE_CODEC", "REQUEST_PEER_LIST", "SEND_FLOWFILES",
+ "RECEIVE_FLOWFILES", "SHUTDOWN" };
//! Respond Code
typedef enum {
- RESERVED = 0,
- // ResponseCode, so that we can indicate a 0 followed by some other bytes
-
- // handshaking properties
- PROPERTIES_OK = 1,
- UNKNOWN_PROPERTY_NAME = 230,
- ILLEGAL_PROPERTY_VALUE = 231,
- MISSING_PROPERTY = 232,
- // transaction indicators
- CONTINUE_TRANSACTION = 10,
- FINISH_TRANSACTION = 11,
- CONFIRM_TRANSACTION = 12, // "Explanation" of this code is the checksum
- TRANSACTION_FINISHED = 13,
- TRANSACTION_FINISHED_BUT_DESTINATION_FULL = 14,
- CANCEL_TRANSACTION = 15,
- BAD_CHECKSUM = 19,
- // data availability indicators
- MORE_DATA = 20,
- NO_MORE_DATA = 21,
- // port state indicators
- UNKNOWN_PORT = 200,
- PORT_NOT_IN_VALID_STATE = 201,
- PORTS_DESTINATION_FULL = 202,
- // authorization
- UNAUTHORIZED = 240,
- // error indicators
- ABORT = 250,
- UNRECOGNIZED_RESPONSE_CODE = 254,
- END_OF_STREAM = 255
+ RESERVED = 0,
+ // ResponseCode, so that we can indicate a 0 followed by some other bytes
+
+ // handshaking properties
+ PROPERTIES_OK = 1,
+ UNKNOWN_PROPERTY_NAME = 230,
+ ILLEGAL_PROPERTY_VALUE = 231,
+ MISSING_PROPERTY = 232,
+ // transaction indicators
+ CONTINUE_TRANSACTION = 10,
+ FINISH_TRANSACTION = 11,
+ CONFIRM_TRANSACTION = 12, // "Explanation" of this code is the checksum
+ TRANSACTION_FINISHED = 13,
+ TRANSACTION_FINISHED_BUT_DESTINATION_FULL = 14,
+ CANCEL_TRANSACTION = 15,
+ BAD_CHECKSUM = 19,
+ // data availability indicators
+ MORE_DATA = 20,
+ NO_MORE_DATA = 21,
+ // port state indicators
+ UNKNOWN_PORT = 200,
+ PORT_NOT_IN_VALID_STATE = 201,
+ PORTS_DESTINATION_FULL = 202,
+ // authorization
+ UNAUTHORIZED = 240,
+ // error indicators
+ ABORT = 250,
+ UNRECOGNIZED_RESPONSE_CODE = 254,
+ END_OF_STREAM = 255
} RespondCode;
//! Respond Code Class
@@ -175,30 +170,26 @@ typedef struct {
} RespondCodeContext;
//! Respond Code Context
-static RespondCodeContext respondCodeContext[] =
-{
- {RESERVED, "Reserved for Future Use", false},
- {PROPERTIES_OK, "Properties OK", false},
- {UNKNOWN_PROPERTY_NAME, "Unknown Property Name", true},
- {ILLEGAL_PROPERTY_VALUE, "Illegal Property Value", true},
- {MISSING_PROPERTY, "Missing Property", true},
- {CONTINUE_TRANSACTION, "Continue Transaction", false},
- {FINISH_TRANSACTION, "Finish Transaction", false},
- {CONFIRM_TRANSACTION, "Confirm Transaction", true},
- {TRANSACTION_FINISHED, "Transaction Finished", false},
- {TRANSACTION_FINISHED_BUT_DESTINATION_FULL, "Transaction Finished But Destination is Full", false},
- {CANCEL_TRANSACTION, "Cancel Transaction", true},
- {BAD_CHECKSUM, "Bad Checksum", false},
- {MORE_DATA, "More Data Exists", false},
- {NO_MORE_DATA, "No More Data Exists", false},
- {UNKNOWN_PORT, "Unknown Port", false},
- {PORT_NOT_IN_VALID_STATE, "Port Not in a Valid State", true},
- {PORTS_DESTINATION_FULL, "Port's Destination is Full", false},
- {UNAUTHORIZED, "User Not Authorized", true},
- {ABORT, "Abort", true},
- {UNRECOGNIZED_RESPONSE_CODE, "Unrecognized Response Code", false},
- {END_OF_STREAM, "End of Stream", false}
-};
+static RespondCodeContext respondCodeContext[] = { { RESERVED,
+ "Reserved for Future Use", false }, { PROPERTIES_OK, "Properties OK",
+ false }, { UNKNOWN_PROPERTY_NAME, "Unknown Property Name", true }, {
+ ILLEGAL_PROPERTY_VALUE, "Illegal Property Value", true }, {
+ MISSING_PROPERTY, "Missing Property", true }, { CONTINUE_TRANSACTION,
+ "Continue Transaction", false }, { FINISH_TRANSACTION,
+ "Finish Transaction", false }, { CONFIRM_TRANSACTION,
+ "Confirm Transaction", true }, { TRANSACTION_FINISHED,
+ "Transaction Finished", false }, {
+ TRANSACTION_FINISHED_BUT_DESTINATION_FULL,
+ "Transaction Finished But Destination is Full", false }, {
+ CANCEL_TRANSACTION, "Cancel Transaction", true }, { BAD_CHECKSUM,
+ "Bad Checksum", false }, { MORE_DATA, "More Data Exists", false }, {
+ NO_MORE_DATA, "No More Data Exists", false }, { UNKNOWN_PORT,
+ "Unknown Port", false }, { PORT_NOT_IN_VALID_STATE,
+ "Port Not in a Valid State", true }, { PORTS_DESTINATION_FULL,
+ "Port's Destination is Full", false }, { UNAUTHORIZED,
+ "User Not Authorized", true }, { ABORT, "Abort", true }, {
+ UNRECOGNIZED_RESPONSE_CODE, "Unrecognized Response Code", false }, {
+ END_OF_STREAM, "End of Stream", false } };
//! Respond Code Sequence Pattern
static const uint8_t CODE_SEQUENCE_VALUE_1 = (uint8_t) 'R';
@@ -209,94 +200,92 @@ static const uint8_t CODE_SEQUENCE_VALUE_2 = (uint8_t) 'C';
* Protocol.
*/
typedef enum {
- /**
- * Boolean value indicating whether or not the contents of a FlowFile should
- * be GZipped when transferred.
- */
- GZIP,
- /**
- * The unique identifier of the port to communicate with
- */
- PORT_IDENTIFIER,
- /**
- * Indicates the number of milliseconds after the request was made that the
- * client will wait for a response. If no response has been received by the
- * time this value expires, the server can move on without attempting to
- * service the request because the client will have already disconnected.
- */
- REQUEST_EXPIRATION_MILLIS,
- /**
- * The preferred number of FlowFiles that the server should send to the
- * client when pulling data. This property was introduced in version 5 of
- * the protocol.
- */
- BATCH_COUNT,
- /**
- * The preferred number of bytes that the server should send to the client
- * when pulling data. This property was introduced in version 5 of the
- * protocol.
- */
- BATCH_SIZE,
- /**
- * The preferred amount of time that the server should send data to the
- * client when pulling data. This property was introduced in version 5 of
- * the protocol. Value is in milliseconds.
- */
- BATCH_DURATION,
- MAX_HANDSHAKE_PROPERTY
+ /**
+ * Boolean value indicating whether or not the contents of a FlowFile should
+ * be GZipped when transferred.
+ */
+ GZIP,
+ /**
+ * The unique identifier of the port to communicate with
+ */
+ PORT_IDENTIFIER,
+ /**
+ * Indicates the number of milliseconds after the request was made that the
+ * client will wait for a response. If no response has been received by the
+ * time this value expires, the server can move on without attempting to
+ * service the request because the client will have already disconnected.
+ */
+ REQUEST_EXPIRATION_MILLIS,
+ /**
+ * The preferred number of FlowFiles that the server should send to the
+ * client when pulling data. This property was introduced in version 5 of
+ * the protocol.
+ */
+ BATCH_COUNT,
+ /**
+ * The preferred number of bytes that the server should send to the client
+ * when pulling data. This property was introduced in version 5 of the
+ * protocol.
+ */
+ BATCH_SIZE,
+ /**
+ * The preferred amount of time that the server should send data to the
+ * client when pulling data. This property was introduced in version 5 of
+ * the protocol. Value is in milliseconds.
+ */
+ BATCH_DURATION, MAX_HANDSHAKE_PROPERTY
} HandshakeProperty;
//! HandShakeProperty Str
-static const char *HandShakePropertyStr[MAX_HANDSHAKE_PROPERTY] =
-{
- /**
- * Boolean value indicating whether or not the contents of a FlowFile should
- * be GZipped when transferred.
- */
- "GZIP",
- /**
- * The unique identifier of the port to communicate with
- */
- "PORT_IDENTIFIER",
- /**
- * Indicates the number of milliseconds after the request was made that the
- * client will wait for a response. If no response has been received by the
- * time this value expires, the server can move on without attempting to
- * service the request because the client will have already disconnected.
- */
- "REQUEST_EXPIRATION_MILLIS",
- /**
- * The preferred number of FlowFiles that the server should send to the
- * client when pulling data. This property was introduced in version 5 of
- * the protocol.
- */
- "BATCH_COUNT",
- /**
- * The preferred number of bytes that the server should send to the client
- * when pulling data. This property was introduced in version 5 of the
- * protocol.
- */
- "BATCH_SIZE",
- /**
- * The preferred amount of time that the server should send data to the
- * client when pulling data. This property was introduced in version 5 of
- * the protocol. Value is in milliseconds.
- */
- "BATCH_DURATION"
-};
+static const char *HandShakePropertyStr[MAX_HANDSHAKE_PROPERTY] = {
+/**
+ * Boolean value indicating whether or not the contents of a FlowFile should
+ * be GZipped when transferred.
+ */
+"GZIP",
+/**
+ * The unique identifier of the port to communicate with
+ */
+"PORT_IDENTIFIER",
+/**
+ * Indicates the number of milliseconds after the request was made that the
+ * client will wait for a response. If no response has been received by the
+ * time this value expires, the server can move on without attempting to
+ * service the request because the client will have already disconnected.
+ */
+"REQUEST_EXPIRATION_MILLIS",
+/**
+ * The preferred number of FlowFiles that the server should send to the
+ * client when pulling data. This property was introduced in version 5 of
+ * the protocol.
+ */
+"BATCH_COUNT",
+/**
+ * The preferred number of bytes that the server should send to the client
+ * when pulling data. This property was introduced in version 5 of the
+ * protocol.
+ */
+"BATCH_SIZE",
+/**
+ * The preferred amount of time that the server should send data to the
+ * client when pulling data. This property was introduced in version 5 of
+ * the protocol. Value is in milliseconds.
+ */
+"BATCH_DURATION" };
class Site2SiteClientProtocol;
//! Transaction Class
-class Transaction
-{
+class Transaction {
friend class Site2SiteClientProtocol;
public:
//! Constructor
/*!
* Create a new transaction
*/
- Transaction(TransferDirection direction) {
+ explicit Transaction(TransferDirection direction,
+ CRCStream<Site2SitePeer> &stream) :
+ crcStream(std::move(stream)) {
_state = TRANSACTION_STARTED;
_direction = direction;
_dataAvailable = false;
@@ -311,48 +300,50 @@ public:
_uuidStr = uuidStr;
}
//! Destructor
- virtual ~Transaction()
- {
+ virtual ~Transaction() {
}
//! getUUIDStr
- std::string getUUIDStr()
- {
+ std::string getUUIDStr() {
return _uuidStr;
}
//! getState
- TransactionState getState()
- {
+ TransactionState getState() {
return _state;
}
//! isDataAvailable
- bool isDataAvailable()
- {
+ bool isDataAvailable() {
return _dataAvailable;
}
//! setDataAvailable()
- void setDataAvailable(bool value)
- {
+ void setDataAvailable(bool value) {
_dataAvailable = value;
}
//! getDirection
- TransferDirection getDirection()
- {
+ TransferDirection getDirection() {
return _direction;
}
//! getCRC
- long getCRC()
- {
- return _crc.getCRC();
+ long getCRC() {
+ return crcStream.getCRC();
}
//! updateCRC
- void updateCRC(uint8_t *buffer, uint32_t length)
- {
- _crc.update(buffer, length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+ crcStream.updateCRC(buffer, length);
}
+ CRCStream<Site2SitePeer> &getStream() {
+ return crcStream;
+ }
+
+
+ Transaction(const Transaction &parent) = delete;
+ Transaction &operator=(const Transaction &parent) = delete;
+
protected:
private:
+
+ CRCStream<Site2SitePeer> crcStream;
//! Transaction State
TransactionState _state;
//! Transaction Direction
@@ -367,21 +358,14 @@ private:
int _transfers;
//! Number of content bytes
uint64_t _bytes;
- //! CRC32
- CRC32 _crc;
- // Prevent default copy constructor and assignment operation
- // Only support pass by reference or pointer
- Transaction(const Transaction &parent);
- Transaction &operator=(const Transaction &parent);
};
/**
* Represents a piece of data that is to be sent to or that was received from a
* NiFi instance.
*/
-class DataPacket
-{
+class DataPacket {
public:
DataPacket(Site2SiteClientProtocol *protocol, Transaction *transaction,
std::map<std::string, std::string> attributes) {
@@ -397,17 +381,16 @@ public:
};
//! Site2SiteClientProtocol Class
-class Site2SiteClientProtocol
-{
+class Site2SiteClientProtocol {
public:
//! Constructor
/*!
* Create a new control protocol
*/
Site2SiteClientProtocol(Site2SitePeer *peer) {
- _logger = Logger::getLogger();
- _configure = Configure::getConfigure();
- _peer = peer;
+ logger_ = Logger::getLogger();
+ configure_ = Configure::getConfigure();
+ peer_ = peer;
_batchSize = 0;
_batchCount = 0;
_batchDuration = 0;
@@ -426,34 +409,34 @@ public:
_currentCodecVersionIndex = 0;
}
//! Destructor
- virtual ~Site2SiteClientProtocol()
- {
+ virtual ~Site2SiteClientProtocol() {
}
public:
//! setBatchSize
- void setBatchSize(uint64_t size)
- {
+ void setBatchSize(uint64_t size) {
_batchSize = size;
}
//! setBatchCount
- void setBatchCount(uint64_t count)
- {
+ void setBatchCount(uint64_t count) {
_batchCount = count;
}
//! setBatchDuration
- void setBatchDuration(uint64_t duration)
- {
+ void setBatchDuration(uint64_t duration) {
_batchDuration = duration;
}
//! setTimeOut
- void setTimeOut(uint64_t time)
- {
+ void setTimeOut(uint64_t time) {
_timeOut = time;
- if (_peer)
- _peer->setTimeOut(time);
+ if (peer_)
+ peer_->setTimeOut(time);
}
+
+ void setPeer(Site2SitePeer *peer)
+ {
+ peer_ = peer;
+ }
/**
* Provides a reference to the time out
* @returns timeout
@@ -470,21 +453,18 @@ public:
return _portIdStr;
}
//! setPortId
- void setPortId(uuid_t id)
- {
+ void setPortId(uuid_t id) {
uuid_copy(_portId, id);
char idStr[37];
uuid_unparse_lower(id, idStr);
_portIdStr = idStr;
}
//! getResourceName
- std::string getResourceName()
- {
+ std::string getResourceName() {
return "SocketFlowFileProtocol";
}
//! getCodecResourceName
- std::string getCodecResourceName()
- {
+ std::string getCodecResourceName() {
return "StandardFlowFileCodec";
}
//! bootstrap the protocol to the ready for transaction state by going through the state machine
@@ -510,31 +490,31 @@ public:
//! write respond
int writeRespond(RespondCode code, std::string message);
//! getRespondCodeContext
- RespondCodeContext *getRespondCodeContext(RespondCode code)
- {
- for (unsigned int i = 0; i < sizeof(respondCodeContext)/sizeof(RespondCodeContext); i++)
- {
- if (respondCodeContext[i].code == code)
- {
+ RespondCodeContext *getRespondCodeContext(RespondCode code) {
+ for (unsigned int i = 0;
+ i < sizeof(respondCodeContext) / sizeof(RespondCodeContext);
+ i++) {
+ if (respondCodeContext[i].code == code) {
return &respondCodeContext[i];
}
}
return NULL;
}
//! getPeer
- Site2SitePeer *getPeer()
- {
- return _peer;
+ Site2SitePeer *getPeer() {
+ return peer_;
}
//! Creation of a new transaction, return the transaction ID if success,
//! Return NULL when any error occurs
- Transaction *createTransaction(std::string &transactionID, TransferDirection direction);
+ Transaction *createTransaction(std::string &transactionID,
+ TransferDirection direction);
//! Receive the data packet from the transaction
//! Return false when any error occurs
bool receive(std::string transactionID, DataPacket *packet, bool &eof);
//! Send the data packet from the transaction
//! Return false when any error occurs
- bool send(std::string transactionID, DataPacket *packet, FlowFileRecord *flowFile, ProcessSession *session);
+ bool send(std::string transactionID, DataPacket *packet,
+ FlowFileRecord *flowFile, ProcessSession *session);
//! Confirm the data that was sent or received by comparing CRC32's of the data sent and the data received.
bool confirm(std::string transactionID);
//! Cancel the transaction
@@ -550,22 +530,22 @@ public:
//! deleteTransaction
void deleteTransaction(std::string transactionID);
//! Nest Callback Class for write stream
- class WriteCallback : public OutputStreamCallback
- {
- public:
- WriteCallback(DataPacket *packet)
- : _packet(packet) {}
+ class WriteCallback: public OutputStreamCallback {
+ public:
+ WriteCallback(DataPacket *packet) :
+ _packet(packet) {
+ }
DataPacket *_packet;
void process(std::ofstream *stream) {
uint8_t buffer[8192];
int len = _packet->_size;
- while (len > 0)
- {
+ while (len > 0) {
int size = std::min(len, (int) sizeof(buffer));
- int ret = _packet->_protocol->_peer->readData(buffer, size, &_packet->_transaction->_crc);
- if (ret != size)
- {
- _packet->_protocol->_logger->log_error("Site2Site Receive Flow Size %d Failed %d", size, ret);
+ int ret = _packet->_transaction->getStream().readData(buffer,size);
+ if (ret != size) {
+ _packet->_protocol->logger_->log_error(
+ "Site2Site Receive Flow Size %d Failed %d", size,
+ ret);
break;
}
stream->write((const char *) buffer, size);
@@ -574,26 +554,26 @@ public:
}
};
//! Nest Callback Class for read stream
- class ReadCallback : public InputStreamCallback
- {
- public:
- ReadCallback(DataPacket *packet)
- : _packet(packet) {}
+ class ReadCallback: public InputStreamCallback {
+ public:
+ ReadCallback(DataPacket *packet) :
+ _packet(packet) {
+ }
DataPacket *_packet;
void process(std::ifstream *stream) {
_packet->_size = 0;
uint8_t buffer[8192];
int readSize;
- while (stream->good())
- {
- if (!stream->read((char *)buffer, 8192))
+ while (stream->good()) {
+ if (!stream->read((char *) buffer, 8192))
readSize = stream->gcount();
else
readSize = 8192;
- int ret = _packet->_protocol->_peer->write(buffer, readSize, &_packet->_transaction->_crc);
- if (ret != readSize)
- {
- _packet->_protocol->_logger->log_error("Site2Site Send Flow Size %d Failed %d", readSize, ret);
+ int ret = _packet->_transaction->getStream().writeData(buffer,readSize);
+ if (ret != readSize) {
+ _packet->_protocol->logger_->log_error(
+ "Site2Site Send Flow Size %d Failed %d", readSize,
+ ret);
break;
}
_packet->_size += readSize;
@@ -604,12 +584,13 @@ public:
protected:
private:
+
//! Mutex for protection
std::mutex _mtx;
//! Logger
- Logger *_logger;
+ Logger *logger_;
//! Configure
- Configure *_configure;
+ Configure *configure_;
//! Batch Count
std::atomic<uint64_t> _batchCount;
//! Batch Size
@@ -619,7 +600,7 @@ private:
//! Timeout in msec
std::atomic<uint64_t> _timeOut;
//! Peer Connection
- Site2SitePeer *_peer;
+ Site2SitePeer *peer_;
//! portId
uuid_t _portId;
//! portIDStr