You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2022/10/06 10:53:20 UTC
[nifi-minifi-cpp] branch main updated (9a3616ff1 -> 248953e96)
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 9a3616ff1 MINIFICPP-1933 Fix Lua unit tests on Windows
new bf5d3d25b MINIFICPP-1923 Refactor PutUDP to use asio
new f5b2ec7ff MINIFICPP-1941 Upgrade OPC UA library to version 1.3.3
new e812c3324 MINIFICPP-1943 Fixing memory leak in NetworkInterfaceInfo
new 248953e96 MINIFICPP-1947 add EXCLUDE_FROM_DEFAULT_BUILD to linter
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/ci.yml | 6 +-
CMakeLists.txt | 42 +++----
cmake/BundledOpen62541.cmake | 9 +-
cmake/Extensions.cmake | 11 +-
cmake/MiNiFiOptions.cmake | 3 +-
.../minifi/core/OPCUAServerContainer.py | 2 +-
extensions/opc/include/opc.h | 11 +-
extensions/opc/src/opc.cpp | 64 +++++-----
.../standard-processors/processors/PutUDP.cpp | 94 +++++++--------
extensions/standard-processors/processors/PutUDP.h | 2 -
.../tests/unit/ListenSyslogTests.cpp | 131 +++++++++++++++------
.../tests/unit/ListenTcpTests.cpp | 78 ++++++++++--
.../standard-processors/tests/unit/PutUDPTests.cpp | 118 ++++++++++---------
libminifi/include/utils/NetworkInterfaceInfo.h | 32 ++---
.../include/utils/net/SessionHandlingServer.h | 2 +-
libminifi/src/utils/NetworkInterfaceInfo.cpp | 16 +--
libminifi/src/utils/net/UdpServer.cpp | 2 +-
libminifi/test/Utils.h | 48 ++++++--
thirdparty/open62541/open62541.patch | 49 ++++----
win_build_vs.bat | 4 +-
20 files changed, 418 insertions(+), 306 deletions(-)
[nifi-minifi-cpp] 02/04: MINIFICPP-1941 Upgrade OPC UA library to version 1.3.3
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit f5b2ec7ff12994c39c05891fc06fbf24c5f84cba
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Oct 6 12:39:24 2022 +0200
MINIFICPP-1941 Upgrade OPC UA library to version 1.3.3
Closes #1423
Signed-off-by: Marton Szasz <sz...@apache.org>
---
cmake/BundledOpen62541.cmake | 9 +--
.../minifi/core/OPCUAServerContainer.py | 2 +-
extensions/opc/include/opc.h | 11 ++--
extensions/opc/src/opc.cpp | 64 +++++++++++-----------
thirdparty/open62541/open62541.patch | 49 +++++++----------
5 files changed, 64 insertions(+), 71 deletions(-)
diff --git a/cmake/BundledOpen62541.cmake b/cmake/BundledOpen62541.cmake
index da1369a48..56c75ea98 100644
--- a/cmake/BundledOpen62541.cmake
+++ b/cmake/BundledOpen62541.cmake
@@ -34,16 +34,17 @@ function(use_bundled_open62541 SOURCE_DIR BINARY_DIR)
set(OPEN62541_CMAKE_ARGS ${PASSTHROUGH_CMAKE_ARGS}
"-DCMAKE_INSTALL_PREFIX=${OPEN62541_BYPRODUCT_DIR}"
- -DOPEN62541_VERSION=v1.0
- -DUA_ENABLE_ENCRYPTION=ON)
+ -DOPEN62541_VERSION=v1.3.3
+ -DUA_ENABLE_ENCRYPTION=ON
+ -DUA_FORCE_WERROR=OFF)
append_third_party_passthrough_args(OPEN62541_CMAKE_ARGS "${OPEN62541_CMAKE_ARGS}")
# Build project
ExternalProject_Add(
open62541-external
- URL "https://github.com/open62541/open62541/archive/v1.2.2.tar.gz"
- URL_HASH "SHA256=9b5bfd811ee523be601f11abc514a93c67fe5c6e957cd6c36fe6ea4f28e009bb"
+ URL "https://github.com/open62541/open62541/archive/v1.3.3.tar.gz"
+ URL_HASH "SHA256=52c049c0f107b4cc382c9e480d677a6360cdd96c472f84689af91b423bd108cb"
SOURCE_DIR "${BINARY_DIR}/thirdparty/open62541-src"
PATCH_COMMAND ${PC}
LIST_SEPARATOR % # This is needed for passing semicolon-separated lists
diff --git a/docker/test/integration/minifi/core/OPCUAServerContainer.py b/docker/test/integration/minifi/core/OPCUAServerContainer.py
index 77e4e59e7..ca2cd2861 100644
--- a/docker/test/integration/minifi/core/OPCUAServerContainer.py
+++ b/docker/test/integration/minifi/core/OPCUAServerContainer.py
@@ -31,7 +31,7 @@ class OPCUAServerContainer(Container):
logging.info('Creating and running OPC UA server docker container...')
self.client.containers.run(
- "lordgamez/open62541",
+ "lordgamez/open62541:1.3.3",
detach=True,
name=self.name,
network=self.network.name,
diff --git a/extensions/opc/include/opc.h b/extensions/opc/include/opc.h
index 6c0f9872d..173b7bf04 100644
--- a/extensions/opc/include/opc.h
+++ b/extensions/opc/include/opc.h
@@ -24,6 +24,7 @@
#include <set>
#include <vector>
#include <memory>
+#include <string_view>
#include "open62541/client.h"
#include "logging/Logger.h"
@@ -55,7 +56,7 @@ class Client {
~Client();
NodeData getNodeData(const UA_ReferenceDescription *ref, const std::string& basePath = "");
UA_ReferenceDescription * getNodeReference(UA_NodeId nodeId);
- void traverse(UA_NodeId nodeId, std::function<nodeFoundCallBackFunc> cb, const std::string& basePath = "", uint64_t maxDepth = 0, bool fetchRoot = true);
+ void traverse(UA_NodeId nodeId, const std::function<nodeFoundCallBackFunc>& cb, const std::string& basePath = "", uint64_t maxDepth = 0, bool fetchRoot = true);
bool exists(UA_NodeId nodeId);
UA_StatusCode translateBrowsePathsToNodeIdsRequest(const std::string& path, std::vector<UA_NodeId>& foundNodeIDs, const std::shared_ptr<core::logging::Logger>& logger);
@@ -63,14 +64,14 @@ class Client {
UA_StatusCode update_node(const UA_NodeId nodeId, T value);
template<typename T>
- UA_StatusCode add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, T value, UA_NodeId *receivedNodeId);
+ UA_StatusCode add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, T value, UA_NodeId *receivedNodeId);
- static std::unique_ptr<Client> createClient(std::shared_ptr<core::logging::Logger> logger, const std::string& applicationURI,
+ static std::unique_ptr<Client> createClient(const std::shared_ptr<core::logging::Logger>& logger, const std::string& applicationURI,
const std::vector<char>& certBuffer, const std::vector<char>& keyBuffer,
const std::vector<std::vector<char>>& trustBuffers);
private:
- Client(std::shared_ptr<core::logging::Logger> logger, const std::string& applicationURI,
+ Client(const std::shared_ptr<core::logging::Logger>& logger, const std::string& applicationURI,
const std::vector<char>& certBuffer, const std::vector<char>& keyBuffer,
const std::vector<std::vector<char>>& trustBuffers);
@@ -82,7 +83,7 @@ using ClientPtr = std::unique_ptr<Client>;
struct NodeData {
std::vector<uint8_t> data;
- uint16_t dataTypeID;
+ UA_DataTypeKind dataTypeID;
std::map<std::string, std::string> attributes;
virtual ~NodeData() {
diff --git a/extensions/opc/src/opc.cpp b/extensions/opc/src/opc.cpp
index 71d9f9482..c10bb4766 100644
--- a/extensions/opc/src/opc.cpp
+++ b/extensions/opc/src/opc.cpp
@@ -108,7 +108,7 @@ core::logging::LOG_LEVEL MapOPCLogLevel(UA_LogLevel ualvl) {
* End of internal functions
*/
-Client::Client(std::shared_ptr<core::logging::Logger> logger, const std::string& applicationURI,
+Client::Client(const std::shared_ptr<core::logging::Logger>& logger, const std::string& applicationURI,
const std::vector<char>& certBuffer, const std::vector<char>& keyBuffer,
const std::vector<std::vector<char>>& trustBuffers) {
client_ = UA_Client_new();
@@ -221,7 +221,7 @@ NodeData Client::getNodeData(const UA_ReferenceDescription *ref, const std::stri
}
nodedata.attributes["Browsename"] = browsename;
nodedata.attributes["Full path"] = basePath + "/" + browsename;
- nodedata.dataTypeID = UA_TYPES_COUNT;
+ nodedata.dataTypeID = static_cast<UA_DataTypeKind>(UA_DATATYPEKINDS);
UA_Variant* var = UA_Variant_new();
if (UA_Client_readValueAttribute(client_, ref->nodeId.nodeId, var) == UA_STATUSCODE_GOOD && var->type != nullptr && var->data != nullptr) {
// Because the timestamps are eliminated in readValueAttribute for simplification
@@ -243,7 +243,7 @@ NodeData Client::getNodeData(const UA_ReferenceDescription *ref, const std::stri
nodedata.attributes["Sourcetimestamp"] = source_timestamp;
UA_ReadResponse_clear(&response);
- nodedata.dataTypeID = var->type->typeIndex;
+ nodedata.dataTypeID = static_cast<UA_DataTypeKind>(var->type->typeKind);
nodedata.addVariant(var);
if (var->type->typeName) {
nodedata.attributes["Typename"] = std::string(var->type->typeName);
@@ -276,7 +276,7 @@ UA_ReferenceDescription * Client::getNodeReference(UA_NodeId nodeId) {
return ref;
}
-void Client::traverse(UA_NodeId nodeId, std::function<nodeFoundCallBackFunc> cb, const std::string& basePath, uint64_t maxDepth, bool fetchRoot) {
+void Client::traverse(UA_NodeId nodeId, const std::function<nodeFoundCallBackFunc>& cb, const std::string& basePath, uint64_t maxDepth, bool fetchRoot) {
if (fetchRoot) {
UA_ReferenceDescription *rootRef = getNodeReference(nodeId);
if ((rootRef->nodeClass == UA_NODECLASS_VARIABLE || rootRef->nodeClass == UA_NODECLASS_OBJECT) && rootRef->browseName.name.length > 0) {
@@ -397,16 +397,16 @@ UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& pa
}
template<typename T>
-UA_StatusCode Client::add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, T value, UA_NodeId *receivedNodeId) {
+UA_StatusCode Client::add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, T value, UA_NodeId *receivedNodeId) {
UA_VariableAttributes attr = UA_VariableAttributes_default;
add_value_to_variant(&attr.value, value);
char local[6] = "en-US";
- attr.displayName = UA_LOCALIZEDTEXT(local, const_cast<char*>(browseName.c_str()));
+ attr.displayName = UA_LOCALIZEDTEXT(local, const_cast<char*>(browseName.data()));
UA_StatusCode sc = UA_Client_addVariableNode(client_,
targetNodeId,
parentNodeId,
UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES),
- UA_QUALIFIEDNAME(1, const_cast<char*>(browseName.c_str())),
+ UA_QUALIFIEDNAME(1, const_cast<char*>(browseName.data())),
UA_NODEID_NULL,
attr, receivedNodeId);
UA_Variant_clear(&attr.value);
@@ -422,7 +422,7 @@ UA_StatusCode Client::update_node(const UA_NodeId nodeId, T value) {
return sc;
}
-std::unique_ptr<Client> Client::createClient(std::shared_ptr<core::logging::Logger> logger, const std::string& applicationURI,
+std::unique_ptr<Client> Client::createClient(const std::shared_ptr<core::logging::Logger>& logger, const std::string& applicationURI,
const std::vector<char>& certBuffer, const std::vector<char>& keyBuffer,
const std::vector<std::vector<char>>& trustBuffers) {
try {
@@ -443,16 +443,16 @@ template UA_StatusCode Client::update_node<bool>(const UA_NodeId nodeId, bool va
template UA_StatusCode Client::update_node<const char *>(const UA_NodeId nodeId, const char * value);
template UA_StatusCode Client::update_node<std::string>(const UA_NodeId nodeId, std::string value);
-template UA_StatusCode Client::add_node<int64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, int64_t value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<uint64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, uint64_t value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<int32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, int32_t value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<uint32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, uint32_t value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<float>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, float value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<double>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, double value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<bool>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, bool value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<const char *>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName,
+template UA_StatusCode Client::add_node<int64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, int64_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<uint64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, uint64_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<int32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, int32_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<uint32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, uint32_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<float>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, float value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<double>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, double value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<bool>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName, bool value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<const char *>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName,
const char * value, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<std::string>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName,
+template UA_StatusCode Client::add_node<std::string>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string_view browseName,
std::string value, UA_NodeId *receivedNodeId);
std::set<std::string> stringToOPCDataTypeMapKeys() {
@@ -466,59 +466,59 @@ std::set<std::string> stringToOPCDataTypeMapKeys() {
std::string nodeValue2String(const NodeData& nd) {
std::string ret_val;
switch (nd.dataTypeID) {
- case UA_TYPES_STRING:
- case UA_TYPES_LOCALIZEDTEXT:
- case UA_TYPES_BYTESTRING: {
+ case UA_DATATYPEKIND_STRING:
+ case UA_DATATYPEKIND_LOCALIZEDTEXT:
+ case UA_DATATYPEKIND_BYTESTRING: {
UA_String value = *reinterpret_cast<UA_String *>(nd.var_->data);
ret_val = std::string(reinterpret_cast<const char *>(value.data), value.length);
break;
}
- case UA_TYPES_BOOLEAN:
+ case UA_DATATYPEKIND_BOOLEAN:
bool b;
memcpy(&b, nd.data.data(), sizeof(bool));
ret_val = b ? "True" : "False";
break;
- case UA_TYPES_SBYTE:
+ case UA_DATATYPEKIND_SBYTE:
int8_t i8t;
memcpy(&i8t, nd.data.data(), sizeof(i8t));
ret_val = std::to_string(i8t);
break;
- case UA_TYPES_BYTE:
+ case UA_DATATYPEKIND_BYTE:
uint8_t ui8t;
memcpy(&ui8t, nd.data.data(), sizeof(ui8t));
ret_val = std::to_string(ui8t);
break;
- case UA_TYPES_INT16:
+ case UA_DATATYPEKIND_INT16:
int16_t i16t;
memcpy(&i16t, nd.data.data(), sizeof(i16t));
ret_val = std::to_string(i16t);
break;
- case UA_TYPES_UINT16:
+ case UA_DATATYPEKIND_UINT16:
uint16_t ui16t;
memcpy(&ui16t, nd.data.data(), sizeof(ui16t));
ret_val = std::to_string(ui16t);
break;
- case UA_TYPES_INT32:
+ case UA_DATATYPEKIND_INT32:
int32_t i32t;
memcpy(&i32t, nd.data.data(), sizeof(i32t));
ret_val = std::to_string(i32t);
break;
- case UA_TYPES_UINT32:
+ case UA_DATATYPEKIND_UINT32:
uint32_t ui32t;
memcpy(&ui32t, nd.data.data(), sizeof(ui32t));
ret_val = std::to_string(ui32t);
break;
- case UA_TYPES_INT64:
+ case UA_DATATYPEKIND_INT64:
int64_t i64t;
memcpy(&i64t, nd.data.data(), sizeof(i64t));
ret_val = std::to_string(i64t);
break;
- case UA_TYPES_UINT64:
+ case UA_DATATYPEKIND_UINT64:
uint64_t ui64t;
memcpy(&ui64t, nd.data.data(), sizeof(ui64t));
ret_val = std::to_string(ui64t);
break;
- case UA_TYPES_FLOAT:
+ case UA_DATATYPEKIND_FLOAT:
if (sizeof(float) == 4 && std::numeric_limits<float>::is_iec559) {
float f;
memcpy(&f, nd.data.data(), sizeof(float));
@@ -527,7 +527,7 @@ std::string nodeValue2String(const NodeData& nd) {
throw OPCException(GENERAL_EXCEPTION, "Float is non-standard on this system, OPC data cannot be extracted!");
}
break;
- case UA_TYPES_DOUBLE:
+ case UA_DATATYPEKIND_DOUBLE:
if (sizeof(double) == 8 && std::numeric_limits<double>::is_iec559) {
double d;
memcpy(&d, nd.data.data(), sizeof(double));
@@ -536,7 +536,7 @@ std::string nodeValue2String(const NodeData& nd) {
throw OPCException(GENERAL_EXCEPTION, "Double is non-standard on this system, OPC data cannot be extracted!");
}
break;
- case UA_TYPES_DATETIME: {
+ case UA_DATATYPEKIND_DATETIME: {
UA_DateTime dt;
memcpy(&dt, nd.data.data(), sizeof(UA_DateTime));
ret_val = opc::OPCDateTime2String(dt);
diff --git a/thirdparty/open62541/open62541.patch b/thirdparty/open62541/open62541.patch
index 17a5ab172..3ad611fe0 100644
--- a/thirdparty/open62541/open62541.patch
+++ b/thirdparty/open62541/open62541.patch
@@ -1,27 +1,18 @@
diff --git a/CMakeLists.txt b/CMakeLists.txt
-index 9184b943..44ae77cf 100644
+index 1934374e..4babdd20 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
-@@ -7,7 +7,7 @@ endif()
-
+@@ -11,7 +11,7 @@ endif()
+
string(TOLOWER "${CMAKE_BUILD_TYPE}" BUILD_TYPE_LOWER_CASE)
-
+
-set(CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/tools/cmake")
+list(APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/tools/cmake")
- find_package(PythonInterp REQUIRED)
+ find_package(Python3 REQUIRED)
+ set(PYTHON_EXECUTABLE ${Python3_EXECUTABLE})
find_package(Git)
- include(AssignSourceGroup)
-@@ -526,7 +526,7 @@ if(NOT UA_FORCE_CPP AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID}" ST
- check_add_cc_flag("-Wall") # Warnings
- check_add_cc_flag("-Wextra") # More warnings
- check_add_cc_flag("-Wpedantic") # Standard compliance
-- check_add_cc_flag("-Werror") # All warnings are errors
-+ # check_add_cc_flag("-Werror") # All warnings are errors
-
- check_add_cc_flag("-Wno-static-in-inline") # Clang doesn't like the use of static inline methods inside static inline methods
- check_add_cc_flag("-Wno-overlength-strings") # May happen in the nodeset compiler when complex values are directly encoded
-@@ -576,17 +576,17 @@ if(NOT UA_FORCE_CPP AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID}" ST
-
+@@ -721,17 +721,17 @@ if(NOT UA_FORCE_CPP AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID}" ST
+
# IPO requires too much memory for unit tests
# GCC docu recommends to compile all files with the same options, therefore ignore it completely
- if(NOT UA_BUILD_UNIT_TESTS AND NOT DEFINED CMAKE_INTERPROCEDURAL_OPTIMIZATION)
@@ -35,17 +26,17 @@ index 9184b943..44ae77cf 100644
- endif()
- endif()
- endif()
-+ # if(NOT UA_BUILD_UNIT_TESTS AND NOT DEFINED CMAKE_INTERPROCEDURAL_OPTIMIZATION)
-+ # # needed to check if IPO is supported (check needs cmake > 3.9)
-+ # if("${CMAKE_VERSION}" VERSION_GREATER 3.9)
-+ # cmake_policy(SET CMP0069 NEW) # needed as long as required cmake < 3.9
-+ # include(CheckIPOSupported)
-+ # check_ipo_supported(RESULT CC_HAS_IPO) # Inter Procedural Optimization / Link Time Optimization (should be same as -flto)
-+ # if(CC_HAS_IPO)
-+ # set(CMAKE_INTERPROCEDURAL_OPTIMIZATION ON)
-+ # endif()
-+ # endif()
-+ # endif()
++ # if(NOT UA_BUILD_UNIT_TESTS AND NOT DEFINED CMAKE_INTERPROCEDURAL_OPTIMIZATION)
++ # # needed to check if IPO is supported (check needs cmake > 3.9)
++ # if("${CMAKE_VERSION}" VERSION_GREATER 3.9)
++ # cmake_policy(SET CMP0069 NEW) # needed as long as required cmake < 3.9
++ # include(CheckIPOSupported)
++ # check_ipo_supported(RESULT CC_HAS_IPO) # Inter Procedural Optimization / Link Time Optimization (should be same as -flto)
++ # if(CC_HAS_IPO)
++ # set(CMAKE_INTERPROCEDURAL_OPTIMIZATION ON)
++ # endif()
++ # endif()
++ # endif()
endif()
-
+
if(UA_ENABLE_AMALGAMATION)
[nifi-minifi-cpp] 04/04: MINIFICPP-1947 add EXCLUDE_FROM_DEFAULT_BUILD to linter
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 248953e9628528ee7df13b0280a1b38b9b0d75b9
Author: Martin Zink <ma...@apache.org>
AuthorDate: Thu Oct 6 12:49:15 2022 +0200
MINIFICPP-1947 add EXCLUDE_FROM_DEFAULT_BUILD to linter
Currently we build the whole solution on windows which linters are part
of. MINIFICPP-1575 intruduced an ENABLE_LINTER cmake option to disable
linter, so it can be built on windows without python. But on the CI we
currently run the linters twice (once implicitly in the build step, and
once again explicitly by the linter step, this also means a linter
failure shows up as a build failure in the CI which can be confusing at
a glance)
If we set the EXCLUDE_FROM_DEFAULT_BUILD target option on all linter
targets, the linters wont be built implicitly by the build step, this
also means we can rid of the ENABLE_LINTER cmake option.
Closes #1427
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.github/workflows/ci.yml | 6 +++---
CMakeLists.txt | 42 +++++++++++++++++++++---------------------
cmake/Extensions.cmake | 11 +++++------
cmake/MiNiFiOptions.cmake | 3 +--
win_build_vs.bat | 4 +---
5 files changed, 31 insertions(+), 35 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6e5e442c6..1aaf77db3 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -73,7 +73,7 @@ jobs:
run: cd ..\b && ctest --timeout 300 --parallel %NUMBER_OF_PROCESSORS% -C Release --output-on-failure
shell: cmd
- name: linter
- run: cd ..\b && msbuild -maxCpuCount linter.vcxproj
+ run: cd ..\b && msbuild /verbosity:quiet -maxCpuCount linter.vcxproj
shell: cmd
ubuntu_20_04:
name: "ubuntu-20.04"
@@ -142,7 +142,7 @@ jobs:
export CXXFLAGS="${CXXFLAGS} -stdlib=libc++"
export LDFLAGS="${LDFLAGS} -stdlib=libc++"
cmake -DUSE_SHARED_LIBS=ON -DCMAKE_BUILD_TYPE=Release -DCI_BUILD=ON -DSTRICT_GSL_CHECKS=AUDIT -DFAIL_ON_WARNINGS=ON -DENABLE_AWS=ON -DENABLE_AZURE=ON -DENABLE_BUSTACHE=ON -DENABLE_COAP=ON \
- -DENABLE_ENCRYPT_CONFIG=ON -DENABLE_GPS=ON -DENABLE_JNI=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_LINTER=ON -DENABLE_MQTT=ON -DENABLE_NANOFI=ON -DENABLE_OPC=ON -DENABLE_OPENCV=ON \
+ -DENABLE_ENCRYPT_CONFIG=ON -DENABLE_GPS=ON -DENABLE_JNI=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_NANOFI=ON -DENABLE_OPC=ON -DENABLE_OPENCV=ON \
-DENABLE_OPENWSMAN=ON -DENABLE_OPS=ON -DENABLE_PCAP=ON -DENABLE_PYTHON=ON -DENABLE_SENSORS=ON -DENABLE_SFTP=ON -DENABLE_SQL=ON -DENABLE_SYSTEMD=ON -DENABLE_TENSORFLOW=OFF \
-DENABLE_USB_CAMERA=ON -DENABLE_SCRIPTING=ON -DENABLE_LUA_SCRIPTING=ON -DENABLE_KUBERNETES=ON -DENABLE_GCP=ON -DENABLE_PROCFS=ON -DENABLE_PROMETHEUS=ON -DENABLE_ELASTICSEARCH=ON \
-DCMAKE_EXPORT_COMPILE_COMMANDS=ON ..
@@ -196,7 +196,7 @@ jobs:
run: |
if [ -d ~/.ccache ]; then mv ~/.ccache .; fi
mkdir build && cd build && cmake -DUSE_SHARED_LIBS=ON -DCI_BUILD=ON -DSTRICT_GSL_CHECKS=AUDIT -DFAIL_ON_WARNINGS=ON -DENABLE_AWS=ON -DENABLE_AZURE=ON -DENABLE_COAP=ON \
- -DENABLE_ENCRYPT_CONFIG=ON -DENABLE_GPS=ON -DENABLE_JNI=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_LINTER=ON -DENABLE_MQTT=ON -DENABLE_NANOFI=ON -DENABLE_OPC=ON -DENABLE_OPENCV=ON \
+ -DENABLE_ENCRYPT_CONFIG=ON -DENABLE_GPS=ON -DENABLE_JNI=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_NANOFI=ON -DENABLE_OPC=ON -DENABLE_OPENCV=ON \
-DENABLE_OPENWSMAN=ON -DENABLE_OPS=ON -DENABLE_PYTHON=ON -DENABLE_SENSORS=ON -DENABLE_SFTP=ON -DENABLE_SQL=ON -DENABLE_SYSTEMD=ON -DENABLE_TENSORFLOW=OFF \
-DENABLE_USB_CAMERA=ON -DENABLE_SCRIPTING=ON -DENABLE_LUA_SCRIPTING=ON -DENABLE_KUBERNETES=ON -DENABLE_GCP=ON -DENABLE_PROCFS=ON -DENABLE_PROMETHEUS=ON -DENABLE_ELASTICSEARCH=ON \
-DDOCKER_SKIP_TESTS=OFF -DDOCKER_BUILD_ONLY=ON -DDOCKER_CCACHE_DUMP_LOCATION=$HOME/.ccache .. && make centos
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 7d25f9a4a..cb79627e9 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -621,27 +621,27 @@ include(BuildDocs)
# Create a custom build target that will run the linter.
# Directories have their separate linter targets to be able to use better parallelization
-if (ENABLE_LINTER)
- get_property(extensions GLOBAL PROPERTY EXTENSION-LINTERS)
- set(root_linted_dirs libminifi/include libminifi/src libminifi/test encrypt-config)
- list(TRANSFORM root_linted_dirs PREPEND ${CMAKE_SOURCE_DIR}/)
-
- set(linted_dir_counter 1)
- set(root_linter_target_names "")
-
- foreach(linted_dir ${root_linted_dirs})
- set(linter_target_name "root-linter-${linted_dir_counter}")
- list(APPEND root_linter_target_names ${linter_target_name})
- add_custom_target("${linter_target_name}"
- COMMAND python3 ${CMAKE_SOURCE_DIR}/thirdparty/google-styleguide/run_linter.py -q -i ${linted_dir}
- )
- math(EXPR linted_dir_counter "${linted_dir_counter}+1")
- endforeach()
-
- # Main linter target that depends on every other
- add_custom_target(linter)
- add_dependencies(linter ${root_linter_target_names} ${extensions})
-endif()
+get_property(extensions GLOBAL PROPERTY EXTENSION-LINTERS)
+set(root_linted_dirs libminifi/include libminifi/src libminifi/test encrypt-config)
+list(TRANSFORM root_linted_dirs PREPEND ${CMAKE_SOURCE_DIR}/)
+
+set(linted_dir_counter 1)
+set(root_linter_target_names "")
+
+foreach(linted_dir ${root_linted_dirs})
+ set(linter_target_name "root-linter-${linted_dir_counter}")
+ list(APPEND root_linter_target_names ${linter_target_name})
+ add_custom_target("${linter_target_name}"
+ COMMAND python3 ${CMAKE_SOURCE_DIR}/thirdparty/google-styleguide/run_linter.py -q -i ${linted_dir}
+ )
+ set_target_properties(${linter_target_name} PROPERTIES EXCLUDE_FROM_DEFAULT_BUILD TRUE)
+ math(EXPR linted_dir_counter "${linted_dir_counter}+1")
+endforeach()
+
+# Main linter target that depends on every other
+add_custom_target(linter)
+add_dependencies(linter ${root_linter_target_names} ${extensions})
+set_target_properties(linter PROPERTIES EXCLUDE_FROM_DEFAULT_BUILD TRUE)
if(NOT WIN32)
diff --git a/cmake/Extensions.cmake b/cmake/Extensions.cmake
index 5df9e53fb..da8a17916 100644
--- a/cmake/Extensions.cmake
+++ b/cmake/Extensions.cmake
@@ -102,12 +102,11 @@ function(registerTest dirName)
endfunction(registerTest)
macro(register_extension_linter target-name)
- if(ENABLE_LINTER)
- get_property(extensions GLOBAL PROPERTY EXTENSION-LINTERS)
- set_property(GLOBAL APPEND PROPERTY EXTENSION-LINTERS "${target-name}")
- add_custom_target(${target-name}
- COMMAND python3 ${CMAKE_SOURCE_DIR}/thirdparty/google-styleguide/run_linter.py -q -i ${CMAKE_CURRENT_LIST_DIR}/)
- endif()
+ get_property(extensions GLOBAL PROPERTY EXTENSION-LINTERS)
+ set_property(GLOBAL APPEND PROPERTY EXTENSION-LINTERS "${target-name}")
+ add_custom_target(${target-name}
+ COMMAND python3 ${CMAKE_SOURCE_DIR}/thirdparty/google-styleguide/run_linter.py -q -i ${CMAKE_CURRENT_LIST_DIR}/)
+ set_target_properties(${target-name} PROPERTIES EXCLUDE_FROM_DEFAULT_BUILD TRUE)
endmacro()
function(get_component_name extension-name output)
diff --git a/cmake/MiNiFiOptions.cmake b/cmake/MiNiFiOptions.cmake
index a4429e499..cfca8b213 100644
--- a/cmake/MiNiFiOptions.cmake
+++ b/cmake/MiNiFiOptions.cmake
@@ -29,7 +29,6 @@ function(add_minifi_dependent_option OPTION_NAME OPTION_DESCRIPTION OPTION_VALUE
endfunction()
add_minifi_option(CI_BUILD "Build is used for CI." OFF)
-add_minifi_option(ENABLE_LINTER "Create linter components" ON)
add_minifi_option(SKIP_TESTS "Skips building all tests." OFF)
add_minifi_option(DOCKER_BUILD_ONLY "Disables all targets except docker build scripts. Ideal for systems without an up-to-date compiler." OFF)
add_minifi_option(DOCKER_SKIP_TESTS "Skip building tests in docker image targets." ON)
@@ -128,7 +127,7 @@ endif()
function(get_minifi_docker_options RET_VALUE)
set(MINIFI_DOCKER_OPTIONS_STR ${MINIFI_EXTERNAL_DOCKER_OPTIONS_STR})
foreach(MINIFI_OPTION ${MINIFI_OPTIONS})
- if (MINIFI_OPTION STREQUAL "CI_BUILD" OR MINIFI_OPTION STREQUAL "ENABLE_LINTER" OR MINIFI_OPTION STREQUAL "SKIP_TESTS" OR MINIFI_OPTION STREQUAL "DOCKER_BUILD_ONLY" OR MINIFI_OPTION STREQUAL "DOCKER_SKIP_TESTS")
+ if (MINIFI_OPTION STREQUAL "CI_BUILD" OR MINIFI_OPTION STREQUAL "SKIP_TESTS" OR MINIFI_OPTION STREQUAL "DOCKER_BUILD_ONLY" OR MINIFI_OPTION STREQUAL "DOCKER_SKIP_TESTS")
continue()
endif()
set(MINIFI_DOCKER_OPTIONS_STR "${MINIFI_DOCKER_OPTIONS_STR} -D${MINIFI_OPTION}=${${MINIFI_OPTION}}")
diff --git a/win_build_vs.bat b/win_build_vs.bat
index 88572536d..39ef9b62c 100755
--- a/win_build_vs.bat
+++ b/win_build_vs.bat
@@ -52,7 +52,6 @@ set cpack=OFF
set installer_merge_modules=OFF
set strict_gsl_checks=
set redist=
-set build_linter=OFF
set build_nanofi=OFF
set build_opencv=OFF
set build_prometheus=OFF
@@ -99,7 +98,6 @@ for %%x in (%*) do (
if [%%~x] EQU [/DD] set cmake_build_type=Debug
if [%%~x] EQU [/CI] set "strict_gsl_checks=-DSTRICT_GSL_CHECKS=AUDIT" & set test_custom_wel_provider=ON
if [%%~x] EQU [/NONFREEUCRT] set "redist=-DMSI_REDISTRIBUTE_UCRT_NONASL=ON"
- if [%%~x] EQU [/L] set build_linter=ON
if [%%~x] EQU [/RO] set real_odbc=ON
if [%%~x] EQU [/NINJA] set generator="Ninja"
)
@@ -121,7 +119,7 @@ cmake -G %generator% -A %build_platform% -DINSTALLER_MERGE_MODULES=%installer_me
-DENABLE_MQTT=%enable_mqtt% -DENABLE_OPC=%enable_opc% -DENABLE_OPENWSMAN=%enable_openwsman% -DENABLE_OPS=%enable_ops% -DENABLE_PCAP=%enable_pcap% ^
-DENABLE_SCRIPTING=%enable_scripting% -DENABLE_SENSORS=%enable_sensors% -DENABLE_TENSORFLOW=%enable_tensorflow% -DENABLE_USB_CAMERA=%enable_usb_camera% ^
-DBUILD_ROCKSDB=ON -DFORCE_WINDOWS=ON -DUSE_SYSTEM_UUID=OFF -DDISABLE_LIBARCHIVE=OFF -DEXCLUDE_BOOST=ON -DENABLE_WEL=ON -DFAIL_ON_WARNINGS=OFF -DSKIP_TESTS=%skiptests% ^
- %strict_gsl_checks% %redist% -DENABLE_LINTER=%build_linter% %EXTRA_CMAKE_ARGUMENTS% "%scriptdir%" && %buildcmd%
+ %strict_gsl_checks% %redist% %EXTRA_CMAKE_ARGUMENTS% "%scriptdir%" && %buildcmd%
IF %ERRORLEVEL% NEQ 0 EXIT /b %ERRORLEVEL%
if [%cpack%] EQU [ON] (
cpack -C %cmake_build_type%
[nifi-minifi-cpp] 03/04: MINIFICPP-1943 Fixing memory leak in NetworkInterfaceInfo
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit e812c3324d1c293a399eab0b5537ad6f42c1cc52
Author: Martin Zink <ma...@apache.org>
AuthorDate: Thu Oct 6 12:47:40 2022 +0200
MINIFICPP-1943 Fixing memory leak in NetworkInterfaceInfo
Closes #1425
Signed-off-by: Marton Szasz <sz...@apache.org>
---
libminifi/include/utils/NetworkInterfaceInfo.h | 32 ++++++++++----------------
libminifi/src/utils/NetworkInterfaceInfo.cpp | 16 ++++---------
2 files changed, 16 insertions(+), 32 deletions(-)
diff --git a/libminifi/include/utils/NetworkInterfaceInfo.h b/libminifi/include/utils/NetworkInterfaceInfo.h
index 12ac06211..5d8f7953e 100644
--- a/libminifi/include/utils/NetworkInterfaceInfo.h
+++ b/libminifi/include/utils/NetworkInterfaceInfo.h
@@ -32,12 +32,7 @@ typedef IP_ADAPTER_ADDRESSES_LH IP_ADAPTER_ADDRESSES;
struct ifaddrs;
#endif
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-
-namespace utils {
+namespace org::apache::nifi::minifi::utils {
class NetworkInterfaceInfo {
public:
NetworkInterfaceInfo(NetworkInterfaceInfo&& src) noexcept = default;
@@ -49,17 +44,18 @@ class NetworkInterfaceInfo {
explicit NetworkInterfaceInfo(const struct ifaddrs* ifa);
#endif
NetworkInterfaceInfo& operator=(NetworkInterfaceInfo&& other) noexcept = default;
- const std::string& getName() const noexcept { return name_; }
- bool hasIpV4Address() const noexcept { return !ip_v4_addresses_.empty(); }
- bool hasIpV6Address() const noexcept { return !ip_v6_addresses_.empty(); }
- bool isRunning() const noexcept { return running_; }
- bool isLoopback() const noexcept { return loopback_; }
- const std::vector<std::string>& getIpV4Addresses() const noexcept { return ip_v4_addresses_; }
- const std::vector<std::string>& getIpV6Addresses() const noexcept { return ip_v6_addresses_; }
+ [[nodiscard]] const std::string& getName() const noexcept { return name_; }
+ [[nodiscard]] bool hasIpV4Address() const noexcept { return !ip_v4_addresses_.empty(); }
+ [[nodiscard]] bool hasIpV6Address() const noexcept { return !ip_v6_addresses_.empty(); }
+ [[nodiscard]] bool isRunning() const noexcept { return running_; }
+ [[nodiscard]] bool isLoopback() const noexcept { return loopback_; }
+ [[nodiscard]] const std::vector<std::string>& getIpV4Addresses() const noexcept { return ip_v4_addresses_; }
+ [[nodiscard]] const std::vector<std::string>& getIpV6Addresses() const noexcept { return ip_v6_addresses_; }
// Traverses the ip addresses and merges them together based on the interface name
- static std::vector<NetworkInterfaceInfo> getNetworkInterfaceInfos(std::function<bool(const NetworkInterfaceInfo&)> filter = { [](const NetworkInterfaceInfo&) { return true; } },
- const std::optional<uint32_t> max_interfaces = std::nullopt);
+ static std::vector<NetworkInterfaceInfo> getNetworkInterfaceInfos(
+ const std::function<bool(const NetworkInterfaceInfo&)>& filter = { [](const NetworkInterfaceInfo&) { return true; } },
+ std::optional<uint32_t> max_interfaces = std::nullopt);
private:
void moveAddressesInto(NetworkInterfaceInfo& destination);
@@ -71,8 +67,4 @@ class NetworkInterfaceInfo {
bool loopback_;
static std::shared_ptr<core::logging::Logger> logger_;
};
-} /* namespace utils */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/utils/NetworkInterfaceInfo.cpp b/libminifi/src/utils/NetworkInterfaceInfo.cpp
index 3384c6be5..fc9dc5753 100644
--- a/libminifi/src/utils/NetworkInterfaceInfo.cpp
+++ b/libminifi/src/utils/NetworkInterfaceInfo.cpp
@@ -32,11 +32,7 @@
#include <ifaddrs.h>
#endif
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace utils {
+namespace org::apache::nifi::minifi::utils {
std::shared_ptr<core::logging::Logger> NetworkInterfaceInfo::logger_ = core::logging::LoggerFactory<NetworkInterfaceInfo>::getLogger();
@@ -87,7 +83,7 @@ struct HasName {
};
}
-std::vector<NetworkInterfaceInfo> NetworkInterfaceInfo::getNetworkInterfaceInfos(std::function<bool(const NetworkInterfaceInfo&)> filter,
+std::vector<NetworkInterfaceInfo> NetworkInterfaceInfo::getNetworkInterfaceInfos(const std::function<bool(const NetworkInterfaceInfo&)>& filter,
const std::optional<uint32_t> max_interfaces) {
std::vector<NetworkInterfaceInfo> network_adapters;
#ifdef WIN32
@@ -120,7 +116,7 @@ std::vector<NetworkInterfaceInfo> NetworkInterfaceInfo::getNetworkInterfaceInfos
}
#else
struct ifaddrs* interface_addresses = nullptr;
- auto cleanup = gsl::finally([interface_addresses] { freeifaddrs(interface_addresses); });
+ auto cleanup = gsl::finally([&interface_addresses] { freeifaddrs(interface_addresses); });
if (getifaddrs(&interface_addresses) == -1) {
logger_->log_error("getifaddrs failed: %s", std::strerror(errno));
return network_adapters;
@@ -158,8 +154,4 @@ void NetworkInterfaceInfo::moveAddressesInto(NetworkInterfaceInfo& destination)
move_append(std::move(ip_v6_addresses_), destination.ip_v6_addresses_);
}
-} /* namespace utils */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::utils
[nifi-minifi-cpp] 01/04: MINIFICPP-1923 Refactor PutUDP to use asio
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit bf5d3d25ba33a7e02eb9bee98767073f08e86464
Author: Martin Zink <ma...@apache.org>
AuthorDate: Wed Oct 5 16:45:25 2022 +0200
MINIFICPP-1923 Refactor PutUDP to use asio
MINIFICPP-1939 Enable dual stack listening on ListenTCP and ListenSyslog
PutUDP should iterate through endpoints (both ipv4 and ipv6)
Closes #1412
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.../standard-processors/processors/PutUDP.cpp | 94 +++++++--------
extensions/standard-processors/processors/PutUDP.h | 2 -
.../tests/unit/ListenSyslogTests.cpp | 131 +++++++++++++++------
.../tests/unit/ListenTcpTests.cpp | 78 ++++++++++--
.../standard-processors/tests/unit/PutUDPTests.cpp | 118 ++++++++++---------
.../include/utils/net/SessionHandlingServer.h | 2 +-
libminifi/src/utils/net/UdpServer.cpp | 2 +-
libminifi/test/Utils.h | 48 ++++++--
8 files changed, 307 insertions(+), 168 deletions(-)
diff --git a/extensions/standard-processors/processors/PutUDP.cpp b/extensions/standard-processors/processors/PutUDP.cpp
index 95000011d..56af0f306 100644
--- a/extensions/standard-processors/processors/PutUDP.cpp
+++ b/extensions/standard-processors/processors/PutUDP.cpp
@@ -16,29 +16,20 @@
*/
#include "PutUDP.h"
-#ifdef WIN32
-#ifndef WIN32_LEAN_AND_MEAN
-#define WIN32_LEAN_AND_MEAN
-#endif
-#include <winsock2.h>
-#else
-#include <netdb.h>
-#endif /* WIN32 */
-#include <utility>
-
-#include "range/v3/view/join.hpp"
#include "range/v3/range/conversion.hpp"
#include "utils/gsl.h"
#include "utils/expected.h"
-#include "utils/net/DNS.h"
-#include "utils/net/Socket.h"
#include "core/ProcessContext.h"
#include "core/ProcessSession.h"
#include "core/PropertyBuilder.h"
#include "core/Resource.h"
#include "core/logging/LoggerConfiguration.h"
+#include "asio/ip/udp.hpp"
+
+using asio::ip::udp;
+
namespace org::apache::nifi::minifi::processors {
const core::Property PutUDP::Hostname = core::PropertyBuilder::createProperty("Hostname")
@@ -107,51 +98,54 @@ void PutUDP::onTrigger(core::ProcessContext* context, core::ProcessSession* cons
return;
}
- const auto nonthrowing_sockaddr_ntop = [](const sockaddr* const sa) -> std::string {
- return utils::try_expression([sa] { return utils::net::sockaddr_ntop(sa); }).value_or("(n/a)");
+ asio::io_context io_context;
+
+ const auto resolve_hostname = [&io_context, &hostname, &port]() -> nonstd::expected<udp::resolver::results_type, std::error_code> {
+ udp::resolver resolver(io_context);
+ std::error_code error_code;
+ auto results = resolver.resolve(hostname, port, error_code);
+ if (error_code)
+ return nonstd::make_unexpected(error_code);
+ return results;
};
- const auto debug_log_resolved_names = [&, this](const addrinfo& names) -> decltype(auto) {
- if (logger_->should_log(core::logging::LOG_LEVEL::debug)) {
- std::vector<std::string> names_vector;
- for (const addrinfo* it = &names; it; it = it->ai_next) {
- names_vector.push_back(nonthrowing_sockaddr_ntop(it->ai_addr));
+ const auto send_data_to_endpoint = [&io_context, &data, &logger = this->logger_](const udp::resolver::results_type& resolved_query) -> nonstd::expected<void, std::error_code> {
+ std::error_code error;
+ for (const auto& resolver_entry : resolved_query) {
+ error.clear();
+ udp::socket socket(io_context);
+ socket.open(resolver_entry.endpoint().protocol(), error);
+ if (error) {
+ logger->log_debug("opening %s socket failed due to %s ", resolver_entry.endpoint().protocol() == udp::v4() ? "IPv4" : "IPv6", error.message());
+ continue;
}
- logger_->log_debug("resolved \'%s\' to: %s",
- hostname,
- names_vector | ranges::views::join(',') | ranges::to<std::string>());
+ socket.send_to(asio::buffer(data.buffer), resolver_entry.endpoint(), udp::socket::message_flags{}, error);
+ if (error) {
+ core::logging::LOG_DEBUG(logger) << "sending to endpoint " << resolver_entry.endpoint() << " failed due to " << error.message();
+ continue;
+ }
+ core::logging::LOG_DEBUG(logger) << "sending to endpoint " << resolver_entry.endpoint() << " succeeded";
+ return {};
}
- return names;
+ return nonstd::make_unexpected(error);
+ };
+
+ const auto transfer_to_success = [&session, &flow_file]() -> void {
+ session->transfer(flow_file, Success);
};
- utils::net::resolveHost(hostname.c_str(), port.c_str(), utils::net::IpProtocol::UDP)
- | utils::map(utils::dereference)
- | utils::map(debug_log_resolved_names)
- | utils::flatMap([](const auto& names) { return utils::net::open_socket(names); })
- | utils::flatMap([&, this](utils::net::OpenSocketResult socket_handle_and_selected_name) -> nonstd::expected<void, std::error_code> {
- const auto& [socket_handle, selected_name] = socket_handle_and_selected_name;
- logger_->log_debug("connected to %s", nonthrowing_sockaddr_ntop(selected_name->ai_addr));
-#ifdef WIN32
- const char* const buffer_ptr = reinterpret_cast<const char*>(data.buffer.data());
-#else
- const void* const buffer_ptr = data.buffer.data();
-#endif
- const auto send_result = ::sendto(socket_handle.get(), buffer_ptr, data.buffer.size(), 0, selected_name->ai_addr, selected_name->ai_addrlen);
- logger_->log_trace("sendto returned %ld", static_cast<long>(send_result)); // NOLINT: sendto
- if (send_result == utils::net::SocketError) {
- return nonstd::make_unexpected(utils::net::get_last_socket_error());
- }
- session->transfer(flow_file, Success);
- return {};
- })
- | utils::orElse([&, this](std::error_code ec) {
- gsl_Expects(ec);
- logger_->log_error("%s", ec.message());
- session->transfer(flow_file, Failure);
- });
+ const auto transfer_to_failure = [&session, &flow_file, &logger = this->logger_](std::error_code ec) -> void {
+ gsl_Expects(ec);
+ logger->log_error("%s", ec.message());
+ session->transfer(flow_file, Failure);
+ };
+
+ resolve_hostname()
+ | utils::flatMap(send_data_to_endpoint)
+ | utils::map(transfer_to_success)
+ | utils::orElse(transfer_to_failure);
}
REGISTER_RESOURCE(PutUDP, Processor);
} // namespace org::apache::nifi::minifi::processors
-
diff --git a/extensions/standard-processors/processors/PutUDP.h b/extensions/standard-processors/processors/PutUDP.h
index 5217519fb..cf69fce3d 100644
--- a/extensions/standard-processors/processors/PutUDP.h
+++ b/extensions/standard-processors/processors/PutUDP.h
@@ -58,8 +58,6 @@ class PutUDP final : public core::Processor {
void onTrigger(core::ProcessContext*, core::ProcessSession*) final;
private:
- std::string hostname_;
- std::string port_;
std::shared_ptr<core::logging::Logger> logger_;
};
} // namespace org::apache::nifi::minifi::processors
diff --git a/extensions/standard-processors/tests/unit/ListenSyslogTests.cpp b/extensions/standard-processors/tests/unit/ListenSyslogTests.cpp
index c9b8e472e..302af6c3a 100644
--- a/extensions/standard-processors/tests/unit/ListenSyslogTests.cpp
+++ b/extensions/standard-processors/tests/unit/ListenSyslogTests.cpp
@@ -21,6 +21,7 @@
#include "SingleProcessorTestController.h"
#include "Utils.h"
#include "controllers/SSLContextService.h"
+#include "range/v3/algorithm/contains.hpp"
using ListenSyslog = org::apache::nifi::minifi::processors::ListenSyslog;
@@ -29,6 +30,7 @@ using namespace std::literals::chrono_literals;
namespace org::apache::nifi::minifi::test {
constexpr uint64_t SYSLOG_PORT = 10255;
+constexpr auto local_addresses = {"127.0.0.1", "::ffff:127.0.0.1", "::1"};
struct ValidRFC5424Message {
constexpr ValidRFC5424Message(std::string_view message,
@@ -197,21 +199,10 @@ constexpr std::string_view rfc5424_logger_example_1 = R"(<13>1 2022-03-17T10:10:
constexpr std::string_view invalid_syslog = "not syslog";
-void sendUDPPacket(const std::string_view content, uint64_t port) {
- asio::io_context io_context;
- asio::ip::udp::socket socket(io_context);
- asio::ip::udp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), port);
- socket.open(asio::ip::udp::v4());
- std::error_code err;
- socket.send_to(asio::buffer(content, content.size()), remote_endpoint, 0, err);
- REQUIRE(!err);
- socket.close();
-}
-
void check_for_only_basic_attributes(core::FlowFile& flow_file, uint16_t port, std::string_view protocol) {
CHECK(std::to_string(port) == flow_file.getAttribute("syslog.port"));
CHECK(protocol == flow_file.getAttribute("syslog.protocol"));
- CHECK("127.0.0.1" == flow_file.getAttribute("syslog.sender"));
+ CHECK(ranges::contains(local_addresses, flow_file.getAttribute("syslog.sender")));
CHECK(std::nullopt == flow_file.getAttribute("syslog.valid"));
CHECK(std::nullopt == flow_file.getAttribute("syslog.priority"));
@@ -228,7 +219,7 @@ void check_for_only_basic_attributes(core::FlowFile& flow_file, uint16_t port, s
void check_parsed_attributes(const core::FlowFile& flow_file, const ValidRFC5424Message& original_message, uint16_t port, std::string_view protocol) {
CHECK(std::to_string(port) == flow_file.getAttribute("syslog.port"));
CHECK(protocol == flow_file.getAttribute("syslog.protocol"));
- CHECK("127.0.0.1" == flow_file.getAttribute("syslog.sender"));
+ CHECK(ranges::contains(local_addresses, flow_file.getAttribute("syslog.sender")));
CHECK("true" == flow_file.getAttribute("syslog.valid"));
CHECK(original_message.priority_ == flow_file.getAttribute("syslog.priority"));
@@ -247,7 +238,7 @@ void check_parsed_attributes(const core::FlowFile& flow_file, const ValidRFC5424
void check_parsed_attributes(const core::FlowFile& flow_file, const ValidRFC3164Message& original_message, uint16_t port, std::string_view protocol) {
CHECK(std::to_string(port) == flow_file.getAttribute("syslog.port"));
CHECK(protocol == flow_file.getAttribute("syslog.protocol"));
- CHECK("127.0.0.1" == flow_file.getAttribute("syslog.sender"));
+ CHECK(ranges::contains(local_addresses, flow_file.getAttribute("syslog.sender")));
CHECK("true" == flow_file.getAttribute("syslog.valid"));
CHECK(original_message.priority_ == flow_file.getAttribute("syslog.priority"));
@@ -269,19 +260,37 @@ TEST_CASE("ListenSyslog without parsing test", "[ListenSyslog]") {
std::string protocol;
SECTION("UDP") {
+ asio::ip::udp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::udp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::udp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
protocol = "UDP";
REQUIRE(listen_syslog->setProperty(ListenSyslog::ProtocolProperty, "UDP"));
controller.plan->scheduleProcessor(listen_syslog);
- sendUDPPacket(rfc5424_logger_example_1, SYSLOG_PORT);
- sendUDPPacket(invalid_syslog, SYSLOG_PORT);
+ utils::sendUdpDatagram(rfc5424_logger_example_1, endpoint);
+ utils::sendUdpDatagram(invalid_syslog, endpoint);
}
SECTION("TCP") {
+ asio::ip::tcp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
protocol = "TCP";
REQUIRE(listen_syslog->setProperty(ListenSyslog::ProtocolProperty, "TCP"));
controller.plan->scheduleProcessor(listen_syslog);
- REQUIRE(utils::sendMessagesViaTCP({rfc5424_logger_example_1}, SYSLOG_PORT));
- REQUIRE(utils::sendMessagesViaTCP({invalid_syslog}, SYSLOG_PORT));
+ REQUIRE(utils::sendMessagesViaTCP({rfc5424_logger_example_1}, endpoint));
+ REQUIRE(utils::sendMessagesViaTCP({invalid_syslog}, endpoint));
}
std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>> result;
REQUIRE(controller.triggerUntil({{ListenSyslog::Success, 2}}, result, 300ms, 50ms));
@@ -303,25 +312,43 @@ TEST_CASE("ListenSyslog with parsing test", "[ListenSyslog]") {
std::string protocol;
SECTION("UDP") {
+ asio::ip::udp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::udp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::udp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
protocol = "UDP";
REQUIRE(listen_syslog->setProperty(ListenSyslog::ProtocolProperty, "UDP"));
controller.plan->scheduleProcessor(listen_syslog);
std::this_thread::sleep_for(100ms);
- sendUDPPacket(rfc5424_doc_example_1.unparsed_, SYSLOG_PORT);
- sendUDPPacket(rfc5424_doc_example_2.unparsed_, SYSLOG_PORT);
- sendUDPPacket(rfc5424_doc_example_3.unparsed_, SYSLOG_PORT);
- sendUDPPacket(rfc5424_doc_example_4.unparsed_, SYSLOG_PORT);
-
- sendUDPPacket(rfc3164_doc_example_1.unparsed_, SYSLOG_PORT);
- sendUDPPacket(rfc3164_doc_example_2.unparsed_, SYSLOG_PORT);
- sendUDPPacket(rfc3164_doc_example_3.unparsed_, SYSLOG_PORT);
- sendUDPPacket(rfc3164_doc_example_4.unparsed_, SYSLOG_PORT);
-
- sendUDPPacket(rfc5424_logger_example_1, SYSLOG_PORT);
- sendUDPPacket(invalid_syslog, SYSLOG_PORT);
+ utils::sendUdpDatagram(rfc5424_doc_example_1.unparsed_, endpoint);
+ utils::sendUdpDatagram(rfc5424_doc_example_2.unparsed_, endpoint);
+ utils::sendUdpDatagram(rfc5424_doc_example_3.unparsed_, endpoint);
+ utils::sendUdpDatagram(rfc5424_doc_example_4.unparsed_, endpoint);
+
+ utils::sendUdpDatagram(rfc3164_doc_example_1.unparsed_, endpoint);
+ utils::sendUdpDatagram(rfc3164_doc_example_2.unparsed_, endpoint);
+ utils::sendUdpDatagram(rfc3164_doc_example_3.unparsed_, endpoint);
+ utils::sendUdpDatagram(rfc3164_doc_example_4.unparsed_, endpoint);
+
+ utils::sendUdpDatagram(rfc5424_logger_example_1, endpoint);
+ utils::sendUdpDatagram(invalid_syslog, endpoint);
}
SECTION("TCP") {
+ asio::ip::tcp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
protocol = "TCP";
REQUIRE(listen_syslog->setProperty(ListenSyslog::ProtocolProperty, "TCP"));
controller.plan->scheduleProcessor(listen_syslog);
@@ -329,15 +356,15 @@ TEST_CASE("ListenSyslog with parsing test", "[ListenSyslog]") {
REQUIRE(utils::sendMessagesViaTCP({rfc5424_doc_example_1.unparsed_,
rfc5424_doc_example_2.unparsed_,
rfc5424_doc_example_3.unparsed_,
- rfc5424_doc_example_4.unparsed_}, SYSLOG_PORT));
+ rfc5424_doc_example_4.unparsed_}, endpoint));
REQUIRE(utils::sendMessagesViaTCP({rfc3164_doc_example_1.unparsed_,
rfc3164_doc_example_2.unparsed_,
rfc3164_doc_example_3.unparsed_,
- rfc3164_doc_example_4.unparsed_}, SYSLOG_PORT));
+ rfc3164_doc_example_4.unparsed_}, endpoint));
- REQUIRE(utils::sendMessagesViaTCP({rfc5424_logger_example_1}, SYSLOG_PORT));
- REQUIRE(utils::sendMessagesViaTCP({invalid_syslog}, SYSLOG_PORT));
+ REQUIRE(utils::sendMessagesViaTCP({rfc5424_logger_example_1}, endpoint));
+ REQUIRE(utils::sendMessagesViaTCP({invalid_syslog}, endpoint));
}
std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>> result;
@@ -410,19 +437,37 @@ TEST_CASE("ListenSyslog max queue and max batch size test", "[ListenSyslog]") {
LogTestController::getInstance().setWarn<ListenSyslog>();
SECTION("UDP") {
+ asio::ip::udp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::udp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::udp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
REQUIRE(listen_syslog->setProperty(ListenSyslog::ProtocolProperty, "UDP"));
controller.plan->scheduleProcessor(listen_syslog);
for (auto i = 0; i < 100; ++i) {
- sendUDPPacket(rfc5424_doc_example_1.unparsed_, SYSLOG_PORT);
+ utils::sendUdpDatagram(rfc5424_doc_example_1.unparsed_, endpoint);
}
CHECK(utils::countLogOccurrencesUntil("Queue is full. UDP message ignored.", 50, 300ms, 50ms));
}
SECTION("TCP") {
+ asio::ip::tcp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
REQUIRE(listen_syslog->setProperty(ListenSyslog::ProtocolProperty, "TCP"));
controller.plan->scheduleProcessor(listen_syslog);
for (auto i = 0; i < 100; ++i) {
- REQUIRE(utils::sendMessagesViaTCP({rfc5424_doc_example_1.unparsed_}, SYSLOG_PORT));
+ REQUIRE(utils::sendMessagesViaTCP({rfc5424_doc_example_1.unparsed_}, endpoint));
}
CHECK(utils::countLogOccurrencesUntil("Queue is full. TCP message ignored.", 50, 300ms, 50ms));
}
@@ -435,6 +480,15 @@ TEST_CASE("ListenSyslog max queue and max batch size test", "[ListenSyslog]") {
}
TEST_CASE("Test ListenSyslog via TCP with SSL connection", "[ListenSyslog]") {
+ asio::ip::tcp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), SYSLOG_PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), SYSLOG_PORT);
+ }
const auto listen_syslog = std::make_shared<ListenSyslog>("ListenSyslog");
SingleProcessorTestController controller{listen_syslog};
@@ -454,9 +508,8 @@ TEST_CASE("Test ListenSyslog via TCP with SSL connection", "[ListenSyslog]") {
REQUIRE(listen_syslog->setProperty(ListenSyslog::SSLContextService, "SSLContextService"));
ssl_context_service->enable();
controller.plan->scheduleProcessor(listen_syslog);
- REQUIRE(utils::sendMessagesViaSSL({rfc5424_logger_example_1}, SYSLOG_PORT, minifi::utils::file::concat_path(executable_dir, "resources/ca_cert.crt")));
- REQUIRE(utils::sendMessagesViaSSL({invalid_syslog}, SYSLOG_PORT, minifi::utils::file::concat_path(executable_dir, "/resources/ca_cert.crt")));
-
+ REQUIRE(utils::sendMessagesViaSSL({rfc5424_logger_example_1}, endpoint, minifi::utils::file::concat_path(executable_dir, "resources/ca_cert.crt")));
+ REQUIRE(utils::sendMessagesViaSSL({invalid_syslog}, endpoint, minifi::utils::file::concat_path(executable_dir, "/resources/ca_cert.crt")));
std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>> result;
REQUIRE(controller.triggerUntil({{ListenSyslog::Success, 2}}, result, 300ms, 50ms));
CHECK(controller.plan->getContent(result.at(ListenSyslog::Success)[0]) == rfc5424_logger_example_1);
diff --git a/extensions/standard-processors/tests/unit/ListenTcpTests.cpp b/extensions/standard-processors/tests/unit/ListenTcpTests.cpp
index 559db0c2f..247f2e6ff 100644
--- a/extensions/standard-processors/tests/unit/ListenTcpTests.cpp
+++ b/extensions/standard-processors/tests/unit/ListenTcpTests.cpp
@@ -21,6 +21,7 @@
#include "SingleProcessorTestController.h"
#include "Utils.h"
#include "controllers/SSLContextService.h"
+#include "range/v3/algorithm/contains.hpp"
using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
@@ -32,10 +33,20 @@ constexpr uint64_t PORT = 10254;
void check_for_attributes(core::FlowFile& flow_file) {
CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
- CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+ const auto local_addresses = {"127.0.0.1", "::ffff:127.0.0.1", "::1"};
+ CHECK(ranges::contains(local_addresses, flow_file.getAttribute("tcp.sender")));
}
TEST_CASE("ListenTCP test multiple messages", "[ListenTCP]") {
+ asio::ip::tcp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
const auto listen_tcp = std::make_shared<ListenTCP>("ListenTCP");
SingleProcessorTestController controller{listen_tcp};
@@ -44,8 +55,8 @@ TEST_CASE("ListenTCP test multiple messages", "[ListenTCP]") {
REQUIRE(listen_tcp->setProperty(ListenTCP::MaxBatchSize, "2"));
controller.plan->scheduleProcessor(listen_tcp);
- REQUIRE(utils::sendMessagesViaTCP({"test_message_1"}, PORT));
- REQUIRE(utils::sendMessagesViaTCP({"another_message"}, PORT));
+ REQUIRE(utils::sendMessagesViaTCP({"test_message_1"}, endpoint));
+ REQUIRE(utils::sendMessagesViaTCP({"another_message"}, endpoint));
ProcessorTriggerResult result;
REQUIRE(controller.triggerUntil({{ListenTCP::Success, 2}}, result, 300s, 50ms));
CHECK(controller.plan->getContent(result.at(ListenTCP::Success)[0]) == "test_message_1");
@@ -68,6 +79,15 @@ TEST_CASE("ListenTCP can be rescheduled", "[ListenTCP]") {
}
TEST_CASE("ListenTCP max queue and max batch size test", "[ListenTCP]") {
+ asio::ip::tcp::endpoint endpoint;
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
const auto listen_tcp = std::make_shared<ListenTCP>("ListenTCP");
SingleProcessorTestController controller{listen_tcp};
@@ -79,7 +99,7 @@ TEST_CASE("ListenTCP max queue and max batch size test", "[ListenTCP]") {
controller.plan->scheduleProcessor(listen_tcp);
for (auto i = 0; i < 100; ++i) {
- REQUIRE(utils::sendMessagesViaTCP({"test_message"}, PORT));
+ REQUIRE(utils::sendMessagesViaTCP({"test_message"}, endpoint));
}
CHECK(utils::countLogOccurrencesUntil("Queue is full. TCP message ignored.", 50, 300ms, 50ms));
@@ -110,27 +130,61 @@ TEST_CASE("Test ListenTCP with SSL connection", "[ListenTCP]") {
REQUIRE(controller.plan->setProperty(listen_tcp, ListenTCP::SSLContextService.getName(), "SSLContextService"));
std::vector<std::string> expected_successful_messages;
+ asio::ip::tcp::endpoint endpoint;
+
SECTION("Without client certificate verification") {
SECTION("Client certificate not required, Client Auth set to NONE by default") {
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
}
SECTION("Client certificate not required, but validated if provided") {
REQUIRE(controller.plan->setProperty(listen_tcp, ListenTCP::ClientAuth.getName(), "WANT"));
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
}
ssl_context_service->enable();
controller.plan->scheduleProcessor(listen_tcp);
expected_successful_messages = {"test_message_1", "another_message"};
- for (const auto& message : expected_successful_messages) {
- REQUIRE(utils::sendMessagesViaSSL({message}, PORT, minifi::utils::file::concat_path(executable_dir, "resources/ca_cert.crt")));
+ for (const auto& message: expected_successful_messages) {
+ REQUIRE(utils::sendMessagesViaSSL({message}, endpoint, minifi::utils::file::concat_path(executable_dir, "resources/ca_cert.crt")));
}
}
SECTION("With client certificate provided") {
SECTION("Client certificate required") {
REQUIRE(controller.plan->setProperty(listen_tcp, ListenTCP::ClientAuth.getName(), "REQUIRED"));
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
}
SECTION("Client certificate not required but validated") {
REQUIRE(controller.plan->setProperty(listen_tcp, ListenTCP::ClientAuth.getName(), "WANT"));
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
}
ssl_context_service->enable();
controller.plan->scheduleProcessor(listen_tcp);
@@ -143,16 +197,24 @@ TEST_CASE("Test ListenTCP with SSL connection", "[ListenTCP]") {
expected_successful_messages = {"test_message_1", "another_message"};
for (const auto& message : expected_successful_messages) {
- REQUIRE(utils::sendMessagesViaSSL({message}, PORT, minifi::utils::file::FileUtils::get_executable_dir() + "/resources/ca_cert.crt", ssl_data));
+ REQUIRE(utils::sendMessagesViaSSL({message}, endpoint, minifi::utils::file::FileUtils::get_executable_dir() + "/resources/ca_cert.crt", ssl_data));
}
}
SECTION("Required certificate not provided") {
+ SECTION("sending through IPv4", "[IPv4]") {
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), PORT);
+ }
+ SECTION("sending through IPv6", "[IPv6]") {
+ if (utils::isIPv6Disabled())
+ return;
+ endpoint = asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), PORT);
+ }
REQUIRE(controller.plan->setProperty(listen_tcp, ListenTCP::ClientAuth.getName(), "REQUIRED"));
ssl_context_service->enable();
controller.plan->scheduleProcessor(listen_tcp);
- REQUIRE_FALSE(utils::sendMessagesViaSSL({"test_message_1"}, PORT, minifi::utils::file::concat_path(executable_dir, "/resources/ca_cert.crt")));
+ REQUIRE_FALSE(utils::sendMessagesViaSSL({"test_message_1"}, endpoint, minifi::utils::file::concat_path(executable_dir, "/resources/ca_cert.crt")));
}
ProcessorTriggerResult result;
diff --git a/extensions/standard-processors/tests/unit/PutUDPTests.cpp b/extensions/standard-processors/tests/unit/PutUDPTests.cpp
index 9e8e95247..fd313d928 100644
--- a/extensions/standard-processors/tests/unit/PutUDPTests.cpp
+++ b/extensions/standard-processors/tests/unit/PutUDPTests.cpp
@@ -24,78 +24,59 @@
#include "Catch.h"
#include "PutUDP.h"
#include "core/ProcessContext.h"
-#include "utils/net/DNS.h"
-#include "utils/net/Socket.h"
+#include "utils/net/UdpServer.h"
#include "utils/expected.h"
#include "utils/StringUtils.h"
+using namespace std::literals::chrono_literals;
+
namespace org::apache::nifi::minifi::processors {
namespace {
-struct DatagramListener {
- DatagramListener(const char* const hostname, const char* const port)
- :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::UDP).value()},
- open_socket_{utils::net::open_socket(*resolved_names_)
- | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
- {
- const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
- if (bind_result == utils::net::SocketError) {
- throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
- }
- }
-
- struct ReceiveResult {
- std::string remote_address;
- std::string message;
- };
-
- [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
- ReceiveResult result;
- result.message.resize(max_message_size);
- sockaddr_storage remote_address{};
- socklen_t addrlen = sizeof(remote_address);
- const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
- if (recv_result == utils::net::SocketError) {
- throw std::runtime_error{utils::StringUtils::join_pack("recvfrom: ", utils::net::get_last_socket_error().message())};
- }
- result.message.resize(gsl::narrow<size_t>(recv_result));
- result.remote_address = utils::net::sockaddr_ntop(std::launder(reinterpret_cast<sockaddr*>(&remote_address)));
- return result;
+std::optional<utils::net::Message> tryDequeueWithTimeout(utils::net::UdpServer& listener, std::chrono::milliseconds timeout = 200ms, std::chrono::milliseconds interval = 10ms) {
+ auto start_time = std::chrono::system_clock::now();
+ utils::net::Message result;
+ while (start_time + timeout > std::chrono::system_clock::now()) {
+ if (listener.tryDequeue(result))
+ return result;
+ std::this_thread::sleep_for(interval);
}
-
- std::unique_ptr<addrinfo, utils::net::addrinfo_deleter> resolved_names_;
- utils::net::OpenSocketResult open_socket_;
-};
+ return std::nullopt;
+}
} // namespace
-// Testing the failure relationship is not required, because since UDP in general without guarantees, flow files are always routed to success, unless there is
-// some weird IO error with the content repo.
TEST_CASE("PutUDP", "[putudp]") {
- const auto putudp = std::make_shared<PutUDP>("PutUDP");
+ const auto put_udp = std::make_shared<PutUDP>("PutUDP");
auto random_engine = std::mt19937{std::random_device{}()}; // NOLINT: "Missing space before { [whitespace/braces] [5]"
// most systems use ports 32768 - 65535 as ephemeral ports, so avoid binding to those
const auto port = std::uniform_int_distribution<uint16_t>{10000, 32768 - 1}(random_engine);
- const auto port_str = std::to_string(port);
- test::SingleProcessorTestController controller{putudp};
+ test::SingleProcessorTestController controller{put_udp};
LogTestController::getInstance().setTrace<PutUDP>();
LogTestController::getInstance().setTrace<core::ProcessContext>();
- LogTestController::getInstance().setLevelByClassName(spdlog::level::trace, "org::apache::nifi::minifi::core::ProcessContextExpr");
- putudp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
- putudp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", port_str, "')}"));
+ put_udp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
+ put_udp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", std::to_string(port), "')}"));
- DatagramListener listener{"localhost", port_str.c_str()};
+ utils::net::UdpServer listener{std::nullopt, port, core::logging::LoggerFactory<utils::net::UdpServer>().getLogger()};
+
+ auto server_thread = std::thread([&listener]() { listener.run(); });
+ auto cleanup_server = gsl::finally([&]{
+ listener.stop();
+ server_thread.join();
+ });
{
const char* const message = "first message: hello";
const auto result = controller.trigger(message);
const auto& success_flow_files = result.at(PutUDP::Success);
REQUIRE(success_flow_files.size() == 1);
- REQUIRE(result.at(PutUDP::Failure).empty());
- REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
- auto receive_result = listener.receive();
- REQUIRE(receive_result.message == message);
- REQUIRE(!receive_result.remote_address.empty());
+ CHECK(result.at(PutUDP::Failure).empty());
+ CHECK(controller.plan->getContent(success_flow_files[0]) == message);
+ auto received_message = tryDequeueWithTimeout(listener);
+ REQUIRE(received_message);
+ CHECK(received_message->message_data == message);
+ CHECK(received_message->protocol == utils::net::IpProtocol::UDP);
+ CHECK(!received_message->sender_address.to_string().empty());
}
{
@@ -103,12 +84,39 @@ TEST_CASE("PutUDP", "[putudp]") {
const auto result = controller.trigger(message);
const auto& success_flow_files = result.at(PutUDP::Success);
REQUIRE(success_flow_files.size() == 1);
- REQUIRE(result.at(PutUDP::Failure).empty());
- REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
- auto receive_result = listener.receive();
- REQUIRE(receive_result.message == message);
- REQUIRE(!receive_result.remote_address.empty());
+ CHECK(result.at(PutUDP::Failure).empty());
+ CHECK(controller.plan->getContent(success_flow_files[0]) == message);
+ auto received_message = tryDequeueWithTimeout(listener);
+ REQUIRE(received_message);
+ CHECK(received_message->message_data == message);
+ CHECK(received_message->protocol == utils::net::IpProtocol::UDP);
+ CHECK(!received_message->sender_address.to_string().empty());
}
-}
+ {
+ LogTestController::getInstance().clear();
+ auto message = std::string(65536, 'a');
+ const auto result = controller.trigger(message);
+ const auto& failure_flow_files = result.at(PutUDP::Failure);
+ REQUIRE(failure_flow_files.size() == 1);
+ CHECK(result.at(PutUDP::Success).empty());
+ CHECK(controller.plan->getContent(failure_flow_files[0]) == message);
+ CHECK((LogTestController::getInstance().contains("Message too long")
+ || LogTestController::getInstance().contains("A message sent on a datagram socket was larger than the internal message buffer")));
+ }
+
+ {
+ LogTestController::getInstance().clear();
+ const char* const message = "message for invalid host";
+ controller.plan->setProperty(put_udp, PutUDP::Hostname.getName(), "invalid_hostname");
+ const auto result = controller.trigger(message);
+ const auto& failure_flow_files = result.at(PutUDP::Failure);
+ auto received_message = tryDequeueWithTimeout(listener);
+ CHECK(!received_message);
+ REQUIRE(failure_flow_files.size() == 1);
+ CHECK(result.at(PutUDP::Success).empty());
+ CHECK(controller.plan->getContent(failure_flow_files[0]) == message);
+ CHECK((LogTestController::getInstance().contains("Host not found") || LogTestController::getInstance().contains("No such host is known")));
+ }
+}
} // namespace org::apache::nifi::minifi::processors
diff --git a/libminifi/include/utils/net/SessionHandlingServer.h b/libminifi/include/utils/net/SessionHandlingServer.h
index f716a8758..173fdcb02 100644
--- a/libminifi/include/utils/net/SessionHandlingServer.h
+++ b/libminifi/include/utils/net/SessionHandlingServer.h
@@ -29,7 +29,7 @@ class SessionHandlingServer : public Server {
public:
SessionHandlingServer(std::optional<size_t> max_queue_size, uint16_t port, std::shared_ptr<core::logging::Logger> logger)
: Server(max_queue_size, std::move(logger)),
- acceptor_(io_context_, asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port)) {
+ acceptor_(io_context_, asio::ip::tcp::endpoint(asio::ip::tcp::v6(), port)) {
}
void run() override {
diff --git a/libminifi/src/utils/net/UdpServer.cpp b/libminifi/src/utils/net/UdpServer.cpp
index 490a081e7..6137f2ddf 100644
--- a/libminifi/src/utils/net/UdpServer.cpp
+++ b/libminifi/src/utils/net/UdpServer.cpp
@@ -22,7 +22,7 @@ UdpServer::UdpServer(std::optional<size_t> max_queue_size,
uint16_t port,
std::shared_ptr<core::logging::Logger> logger)
: Server(max_queue_size, std::move(logger)),
- socket_(io_context_, asio::ip::udp::endpoint(asio::ip::udp::v4(), port)) {
+ socket_(io_context_, asio::ip::udp::endpoint(asio::ip::udp::v6(), port)) {
doReceive();
}
diff --git a/libminifi/test/Utils.h b/libminifi/test/Utils.h
index 96389f740..76d8cc533 100644
--- a/libminifi/test/Utils.h
+++ b/libminifi/test/Utils.h
@@ -111,24 +111,47 @@ bool countLogOccurrencesUntil(const std::string& pattern,
return false;
}
-bool sendMessagesViaTCP(const std::vector<std::string_view>& contents, uint64_t port) {
+bool sendMessagesViaTCP(const std::vector<std::string_view>& contents, const asio::ip::tcp::endpoint& remote_endpoint) {
asio::io_context io_context;
asio::ip::tcp::socket socket(io_context);
- asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), port);
socket.connect(remote_endpoint);
std::error_code err;
for (auto& content : contents) {
std::string tcp_message(content);
tcp_message += '\n';
asio::write(socket, asio::buffer(tcp_message, tcp_message.size()), err);
+ if (err) {
+ return false;
+ }
}
- if (err) {
- return false;
- }
- socket.close();
return true;
}
+bool sendUdpDatagram(const asio::const_buffer content, const asio::ip::udp::endpoint& remote_endpoint) {
+ asio::io_context io_context;
+ asio::ip::udp::socket socket(io_context);
+ socket.open(remote_endpoint.protocol());
+ std::error_code err;
+ socket.send_to(content, remote_endpoint, 0, err);
+ return !err;
+}
+
+bool sendUdpDatagram(const gsl::span<std::byte const> content, const asio::ip::udp::endpoint& remote_endpoint) {
+ return sendUdpDatagram(asio::const_buffer(content.begin(), content.size()), remote_endpoint);
+}
+
+bool sendUdpDatagram(const std::string_view content, const asio::ip::udp::endpoint& remote_endpoint) {
+ return sendUdpDatagram(asio::buffer(content), remote_endpoint);
+}
+
+bool isIPv6Disabled() {
+ asio::io_context io_context;
+ std::error_code error_code;
+ asio::ip::tcp::socket socket_tcp(io_context);
+ socket_tcp.connect(asio::ip::tcp::endpoint(asio::ip::address_v6::loopback(), 10), error_code);
+ return error_code.value() == EADDRNOTAVAIL;
+}
+
struct ConnectionTestAccessor {
FIELD_ACCESSOR(queue_);
};
@@ -143,7 +166,10 @@ struct FlowFileQueueTestAccessor {
FIELD_ACCESSOR(queue_);
};
-bool sendMessagesViaSSL(const std::vector<std::string_view>& contents, uint64_t port, const std::string& ca_cert_path, const std::optional<minifi::utils::net::SslData>& ssl_data = std::nullopt) {
+bool sendMessagesViaSSL(const std::vector<std::string_view>& contents,
+ const asio::ip::tcp::endpoint& remote_endpoint,
+ const std::string& ca_cert_path,
+ const std::optional<minifi::utils::net::SslData>& ssl_data = std::nullopt) {
asio::ssl::context ctx(asio::ssl::context::sslv23);
ctx.load_verify_file(ca_cert_path);
if (ssl_data) {
@@ -154,7 +180,6 @@ bool sendMessagesViaSSL(const std::vector<std::string_view>& contents, uint64_t
}
asio::io_context io_context;
asio::ssl::stream<asio::ip::tcp::socket> socket(io_context, ctx);
- asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), port);
asio::error_code err;
socket.lowest_layer().connect(remote_endpoint, err);
if (err) {
@@ -168,11 +193,10 @@ bool sendMessagesViaSSL(const std::vector<std::string_view>& contents, uint64_t
std::string tcp_message(content);
tcp_message += '\n';
asio::write(socket, asio::buffer(tcp_message, tcp_message.size()), err);
+ if (err) {
+ return false;
+ }
}
- if (err) {
- return false;
- }
- socket.lowest_layer().close();
return true;
}