You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2024/06/24 10:29:26 UTC
(nifi-minifi-cpp) branch main updated (7532b4ab1 -> ff43ef4f4)
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 7532b4ab1 MINIFICPP-2363 Verify linux package on other distros using the docker test suite
new 1e9b6159b MINIFICPP-2341 Include custom relationships in C2 manifest for NiFi python processors
new 32f3e4ecc MINIFICPP-2389 Fix auto-terminated relationships in python processors
new a9110fdeb MINIFICPP-2399 Evaluate expr language in GetFile's Directory property
new 3bbaa0789 MINIFICPP-2395 Drop unused CoAP C2 protocol
new 7585d5d21 MINIFICPP-2406 Fix formatting of parameters in some remaining log lines
new ff43ef4f4 MINIFICPP-2042 Deprecate the Both output format in CWEL
The 6 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 +-
C2.md | 36 +--
CMakeLists.txt | 2 +-
LICENSE | 29 --
PROCESSORS.md | 36 +--
README.md | 1 -
Windows.md | 3 +-
bootstrap.sh | 5 -
bootstrap/system_dependency.py | 2 +-
bstrp_functions.sh | 2 -
cmake/BundledLibCoAP.cmake | 85 ------
cmake/DockerConfig.cmake | 1 -
cmake/MiNiFiOptions.cmake | 1 -
conf/minifi.properties | 7 -
docker/test/integration/cluster/ImageStore.py | 4 +-
docker/test/integration/features/python.feature | 14 +
...warder.py => ProcessContextInterfaceChecker.py} | 6 +-
.../processors/SpecialPropertyTypeChecker.py | 1 +
.../python/ProcessContextInterfaceChecker.py | 92 +++++++
encrypt-config/tests/ConfigFileEncryptorTests.cpp | 2 +-
encrypt-config/tests/ConfigFileTests.cpp | 8 +-
encrypt-config/tests/resources/minifi.properties | 7 -
...th-additional-sensitive-props.minifi.properties | 7 -
extensions/coap/CMakeLists.txt | 42 ---
extensions/coap/COAPLoader.cpp | 47 ----
.../coap/controllerservice/CoapConnector.cpp | 74 -----
extensions/coap/controllerservice/CoapConnector.h | 126 ---------
extensions/coap/controllerservice/CoapMessaging.h | 109 --------
extensions/coap/controllerservice/CoapResponse.h | 82 ------
extensions/coap/nanofi/coap_connection.c | 124 ---------
extensions/coap/nanofi/coap_connection.h | 67 -----
extensions/coap/nanofi/coap_functions.c | 231 ----------------
extensions/coap/nanofi/coap_functions.h | 126 ---------
extensions/coap/nanofi/coap_message.c | 29 --
extensions/coap/nanofi/coap_message.h | 52 ----
extensions/coap/nanofi/coap_server.c | 78 ------
extensions/coap/nanofi/coap_server.h | 80 ------
extensions/coap/protocols/CoapC2Protocol.cpp | 299 ---------------------
extensions/coap/protocols/CoapC2Protocol.h | 126 ---------
extensions/coap/server/CoapServer.cpp | 44 ---
extensions/coap/server/CoapServer.h | 238 ----------------
extensions/coap/tests/CMakeLists.txt | 53 ----
extensions/coap/tests/CoapC2VerifyHeartbeat.cpp | 180 -------------
extensions/coap/tests/CoapIntegrationBase.h | 123 ---------
extensions/python/ExecutePythonProcessor.cpp | 10 +-
extensions/python/ExecutePythonProcessor.h | 2 +
extensions/python/PYTHON.md | 4 +
extensions/python/PythonCreator.h | 2 +-
extensions/python/PythonScriptEngine.cpp | 39 +++
extensions/python/PythonScriptEngine.h | 31 ++-
.../pythonprocessors/nifiapi/flowfiletransform.py | 8 +-
.../python/pythonprocessors/nifiapi/properties.py | 49 ++--
.../pythonprocessors/nifiapi/relationship.py | 9 +-
extensions/python/types/PyProcessContext.cpp | 38 +++
extensions/python/types/PyProcessContext.h | 2 +
extensions/rocksdb-repos/FlowFileRepository.cpp | 4 +-
.../standard-processors/processors/GetFile.cpp | 2 +-
.../processors/JoltTransformJSON.cpp | 2 +-
.../standard-processors/tests/CMakeLists.txt | 2 +
.../tests/unit/GetFileTests.cpp | 29 ++
.../tests/unit/ProcessorTests.cpp | 2 +-
.../windows-event-log/ConsumeWindowsEventLog.h | 12 +-
libminifi/include/c2/C2Agent.h | 6 +-
libminifi/include/core/Connectable.h | 1 +
libminifi/include/properties/Configuration.h | 4 -
libminifi/src/Configuration.cpp | 4 -
libminifi/src/c2/C2Agent.cpp | 50 +---
libminifi/src/core/Connectable.cpp | 12 +
libminifi/test/integration/C2NullConfiguration.cpp | 5 +-
.../test/integration/C2PropertiesUpdateTests.cpp | 1 -
libminifi/test/integration/C2RequestClassTest.cpp | 1 -
libminifi/test/integration/C2UpdateAssetTest.cpp | 1 -
.../test/integration/C2VerifyServeResults.cpp | 1 -
.../test/libtest/integration/HTTPIntegrationBase.h | 3 -
.../test/resources/encrypted.minifi.properties | 7 -
msi/WixWin.wsi.in | 17 --
thirdparty/libcoap/libcoap-windows-cmake.patch | 93 -------
win_build_vs.bat | 4 +-
78 files changed, 377 insertions(+), 2767 deletions(-)
delete mode 100644 cmake/BundledLibCoAP.cmake
copy docker/test/integration/minifi/processors/{RotatingForwarder.py => ProcessContextInterfaceChecker.py} (86%)
create mode 100644 docker/test/integration/resources/python/ProcessContextInterfaceChecker.py
delete mode 100644 extensions/coap/CMakeLists.txt
delete mode 100644 extensions/coap/COAPLoader.cpp
delete mode 100644 extensions/coap/controllerservice/CoapConnector.cpp
delete mode 100644 extensions/coap/controllerservice/CoapConnector.h
delete mode 100644 extensions/coap/controllerservice/CoapMessaging.h
delete mode 100644 extensions/coap/controllerservice/CoapResponse.h
delete mode 100644 extensions/coap/nanofi/coap_connection.c
delete mode 100644 extensions/coap/nanofi/coap_connection.h
delete mode 100644 extensions/coap/nanofi/coap_functions.c
delete mode 100644 extensions/coap/nanofi/coap_functions.h
delete mode 100644 extensions/coap/nanofi/coap_message.c
delete mode 100644 extensions/coap/nanofi/coap_message.h
delete mode 100644 extensions/coap/nanofi/coap_server.c
delete mode 100644 extensions/coap/nanofi/coap_server.h
delete mode 100644 extensions/coap/protocols/CoapC2Protocol.cpp
delete mode 100644 extensions/coap/protocols/CoapC2Protocol.h
delete mode 100644 extensions/coap/server/CoapServer.cpp
delete mode 100644 extensions/coap/server/CoapServer.h
delete mode 100644 extensions/coap/tests/CMakeLists.txt
delete mode 100644 extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
delete mode 100644 extensions/coap/tests/CoapIntegrationBase.h
copy docker/test/integration/cluster/LogSource.py => extensions/python/pythonprocessors/nifiapi/relationship.py (79%)
delete mode 100644 thirdparty/libcoap/libcoap-windows-cmake.patch
(nifi-minifi-cpp) 02/06: MINIFICPP-2389 Fix auto-terminated relationships in python processors
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 32f3e4ecc1017a7ccb156bcb4b15fd65be070693
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jun 24 10:34:34 2024 +0200
MINIFICPP-2389 Fix auto-terminated relationships in python processors
Closes #1800
Signed-off-by: Martin Zink <ma...@apache.org>
---
extensions/python/ExecutePythonProcessor.cpp | 2 +-
libminifi/include/core/Connectable.h | 1 +
libminifi/src/core/Connectable.cpp | 12 ++++++++++++
3 files changed, 14 insertions(+), 1 deletion(-)
diff --git a/extensions/python/ExecutePythonProcessor.cpp b/extensions/python/ExecutePythonProcessor.cpp
index 0d364803c..e3b0a3a14 100644
--- a/extensions/python/ExecutePythonProcessor.cpp
+++ b/extensions/python/ExecutePythonProcessor.cpp
@@ -76,7 +76,7 @@ void ExecutePythonProcessor::initalizeThroughScriptEngine() {
}
void ExecutePythonProcessor::onScheduleSharedPtr(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) {
- setAutoTerminatedRelationships(std::vector<core::Relationship>{Original});
+ addAutoTerminatedRelationship(Original);
if (!processor_initialized_) {
loadScript();
python_script_engine_ = createScriptEngine();
diff --git a/libminifi/include/core/Connectable.h b/libminifi/include/core/Connectable.h
index 129bf2ef2..82311720b 100644
--- a/libminifi/include/core/Connectable.h
+++ b/libminifi/include/core/Connectable.h
@@ -62,6 +62,7 @@ class Connectable : public CoreComponent {
std::vector<Relationship> getSupportedRelationships() const;
+ void addAutoTerminatedRelationship(const core::Relationship& relationship);
void setAutoTerminatedRelationships(std::span<const core::Relationship> relationships);
bool isAutoTerminated(const Relationship &relationship);
diff --git a/libminifi/src/core/Connectable.cpp b/libminifi/src/core/Connectable.cpp
index 7e9cdb960..55e0ff27f 100644
--- a/libminifi/src/core/Connectable.cpp
+++ b/libminifi/src/core/Connectable.cpp
@@ -59,6 +59,7 @@ void Connectable::setSupportedRelationships(std::span<const core::RelationshipDe
std::vector<Relationship> Connectable::getSupportedRelationships() const {
std::vector<Relationship> relationships;
+ relationships.reserve(relationships_.size());
for (auto const &item : relationships_) {
relationships.push_back(item.second);
}
@@ -75,6 +76,17 @@ bool Connectable::isSupportedRelationship(const core::Relationship &relationship
return relationships_.contains(relationship.getName());
}
+void Connectable::addAutoTerminatedRelationship(const core::Relationship& relationship) {
+ if (isRunning()) {
+ logger_->log_warn("Can not add processor auto terminated relationship while the process {} is running", name_);
+ return;
+ }
+
+ std::lock_guard<std::mutex> lock(relationship_mutex_);
+
+ auto_terminated_relationships_[relationship.getName()] = relationship;
+}
+
void Connectable::setAutoTerminatedRelationships(std::span<const core::Relationship> relationships) {
if (isRunning()) {
logger_->log_warn("Can not set processor auto terminated relationship while the process {} is running", name_);
(nifi-minifi-cpp) 03/06: MINIFICPP-2399 Evaluate expr language in GetFile's Directory property
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit a9110fdebb9b3d6f7aba686a8bd2a43fc3ebf4cb
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jun 24 10:34:41 2024 +0200
MINIFICPP-2399 Evaluate expr language in GetFile's Directory property
Closes #1812
Signed-off-by: Martin Zink <ma...@apache.org>
---
.../standard-processors/processors/GetFile.cpp | 2 +-
.../standard-processors/tests/CMakeLists.txt | 2 ++
.../tests/unit/GetFileTests.cpp | 29 ++++++++++++++++++++++
3 files changed, 32 insertions(+), 1 deletion(-)
diff --git a/extensions/standard-processors/processors/GetFile.cpp b/extensions/standard-processors/processors/GetFile.cpp
index 318996730..c3f901c35 100644
--- a/extensions/standard-processors/processors/GetFile.cpp
+++ b/extensions/standard-processors/processors/GetFile.cpp
@@ -76,7 +76,7 @@ void GetFile::onSchedule(core::ProcessContext& context, core::ProcessSessionFact
request_.fileFilter = value;
}
- if (auto directory_str = context.getProperty(Directory)) {
+ if (auto directory_str = context.getProperty(Directory, nullptr)) {
if (!utils::file::is_directory(*directory_str)) {
throw Exception(PROCESS_SCHEDULE_EXCEPTION, utils::string::join_pack("Input Directory \"", *directory_str, "\" is not a directory"));
}
diff --git a/extensions/standard-processors/tests/CMakeLists.txt b/extensions/standard-processors/tests/CMakeLists.txt
index ed6d8b3af..cf22f305b 100644
--- a/extensions/standard-processors/tests/CMakeLists.txt
+++ b/extensions/standard-processors/tests/CMakeLists.txt
@@ -34,11 +34,13 @@ FOREACH(testfile ${PROCESSOR_UNIT_TESTS})
target_include_directories(${testfilename} BEFORE PRIVATE "../processors")
target_include_directories(${testfilename} BEFORE PRIVATE ./include)
target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/civetweb/")
+ target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/expression-language/")
createTests("${testfilename}")
target_link_libraries(${testfilename} Catch2WithMain)
target_link_libraries(${testfilename} minifi-expression-language-extensions)
target_link_libraries(${testfilename} minifi-standard-processors)
target_compile_definitions("${testfilename}" PRIVATE JOLT_TESTS_DIR="${jolt_tests_SOURCE_DIR}/jolt-core/src/test/resources/json/shiftr")
+ target_compile_definitions("${testfilename}" PRIVATE TZ_DATA_DIR="${CMAKE_BINARY_DIR}/tzdata")
if (ENABLE_ROCKSDB)
target_link_libraries(${testfilename} minifi-rocksdb-repos)
endif()
diff --git a/extensions/standard-processors/tests/unit/GetFileTests.cpp b/extensions/standard-processors/tests/unit/GetFileTests.cpp
index 17c2625c6..90c9fdf23 100644
--- a/extensions/standard-processors/tests/unit/GetFileTests.cpp
+++ b/extensions/standard-processors/tests/unit/GetFileTests.cpp
@@ -28,6 +28,9 @@
#include "utils/file/FileUtils.h"
#include "unit/TestUtils.h"
#include "unit/ProvenanceTestHelper.h"
+#ifdef WIN32
+#include "impl/expression/Expression.h"
+#endif
using namespace std::literals::chrono_literals;
@@ -292,3 +295,29 @@ TEST_CASE("GetFile sets attributes correctly") {
CHECK(flow_file->getAttribute(minifi::core::SpecialFlowAttribute::FILENAME) == "beta.txt");
}
}
+
+TEST_CASE("GetFile can use expression language in Directory property") {
+#ifdef WIN32
+ minifi::expression::dateSetInstall(TZ_DATA_DIR);
+#endif
+ using minifi::processors::GetFile;
+ LogTestController::getInstance().setTrace<GetFile>();
+
+ const auto get_file = std::make_shared<GetFile>("GetFile");
+ minifi::test::SingleProcessorTestController test_controller(get_file);
+
+ std::filesystem::path base_dir = test_controller.createTempDirectory();
+ auto date_str = date::format("%Y-%m-%d", std::chrono::system_clock::now());
+ auto dir = base_dir/ date_str;
+ std::filesystem::create_directories(dir);
+ get_file->setProperty(GetFile::Directory, base_dir.string() + "/${now():format('%Y-%m-%d')}");
+ minifi::test::utils::putFileToDir(dir, "testfile.txt", "The quick brown fox jumps over the lazy dog\n");
+
+ auto result = test_controller.trigger();
+
+ REQUIRE((result.contains(GetFile::Success) && result.at(GetFile::Success).size() == 1));
+ auto flow_file = result.at(GetFile::Success)[0];
+ CHECK(flow_file->getAttribute(minifi::core::SpecialFlowAttribute::PATH) == (std::filesystem::path(".") / "").string());
+ CHECK(flow_file->getAttribute(minifi::core::SpecialFlowAttribute::ABSOLUTE_PATH) == (dir / "").string());
+ CHECK(flow_file->getAttribute(minifi::core::SpecialFlowAttribute::FILENAME) == "testfile.txt");
+}
(nifi-minifi-cpp) 04/06: MINIFICPP-2395 Drop unused CoAP C2 protocol
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3bbaa0789a142e95edb45a73ad68cd8c620a1495
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jun 24 10:35:01 2024 +0200
MINIFICPP-2395 Drop unused CoAP C2 protocol
Closes #1814
Signed-off-by: Martin Zink <ma...@apache.org>
---
.github/workflows/ci.yml | 6 +-
C2.md | 36 +--
CMakeLists.txt | 2 +-
LICENSE | 29 --
README.md | 1 -
Windows.md | 3 +-
bootstrap.sh | 5 -
bootstrap/system_dependency.py | 2 +-
bstrp_functions.sh | 2 -
cmake/BundledLibCoAP.cmake | 85 ------
cmake/DockerConfig.cmake | 1 -
cmake/MiNiFiOptions.cmake | 1 -
conf/minifi.properties | 7 -
encrypt-config/tests/ConfigFileEncryptorTests.cpp | 2 +-
encrypt-config/tests/ConfigFileTests.cpp | 8 +-
encrypt-config/tests/resources/minifi.properties | 7 -
...th-additional-sensitive-props.minifi.properties | 7 -
extensions/coap/CMakeLists.txt | 42 ---
extensions/coap/COAPLoader.cpp | 47 ----
.../coap/controllerservice/CoapConnector.cpp | 74 -----
extensions/coap/controllerservice/CoapConnector.h | 126 ---------
extensions/coap/controllerservice/CoapMessaging.h | 109 --------
extensions/coap/controllerservice/CoapResponse.h | 82 ------
extensions/coap/nanofi/coap_connection.c | 124 ---------
extensions/coap/nanofi/coap_connection.h | 67 -----
extensions/coap/nanofi/coap_functions.c | 231 ----------------
extensions/coap/nanofi/coap_functions.h | 126 ---------
extensions/coap/nanofi/coap_message.c | 29 --
extensions/coap/nanofi/coap_message.h | 52 ----
extensions/coap/nanofi/coap_server.c | 78 ------
extensions/coap/nanofi/coap_server.h | 80 ------
extensions/coap/protocols/CoapC2Protocol.cpp | 299 ---------------------
extensions/coap/protocols/CoapC2Protocol.h | 126 ---------
extensions/coap/server/CoapServer.cpp | 44 ---
extensions/coap/server/CoapServer.h | 238 ----------------
extensions/coap/tests/CMakeLists.txt | 53 ----
extensions/coap/tests/CoapC2VerifyHeartbeat.cpp | 180 -------------
extensions/coap/tests/CoapIntegrationBase.h | 123 ---------
.../tests/unit/ProcessorTests.cpp | 2 +-
libminifi/include/c2/C2Agent.h | 6 +-
libminifi/include/properties/Configuration.h | 4 -
libminifi/src/Configuration.cpp | 4 -
libminifi/src/c2/C2Agent.cpp | 50 +---
libminifi/test/integration/C2NullConfiguration.cpp | 5 +-
.../test/integration/C2PropertiesUpdateTests.cpp | 1 -
libminifi/test/integration/C2RequestClassTest.cpp | 1 -
libminifi/test/integration/C2UpdateAssetTest.cpp | 1 -
.../test/integration/C2VerifyServeResults.cpp | 1 -
.../test/libtest/integration/HTTPIntegrationBase.h | 3 -
.../test/resources/encrypted.minifi.properties | 7 -
msi/WixWin.wsi.in | 17 --
thirdparty/libcoap/libcoap-windows-cmake.patch | 93 -------
win_build_vs.bat | 4 +-
53 files changed, 29 insertions(+), 2704 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7b689287e..8529881ec 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -25,7 +25,6 @@ jobs:
-DENABLE_BUSTACHE=OFF
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
- -DENABLE_COAP=OFF
-DENABLE_CONTROLLER=ON
-DENABLE_ELASTICSEARCH=ON
-DENABLE_ENCRYPT_CONFIG=ON
@@ -138,7 +137,6 @@ jobs:
-DENABLE_BUSTACHE=OFF
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
- -DENABLE_COAP=OFF
-DENABLE_CONTROLLER=ON
-DENABLE_COVERAGE=
-DENABLE_ELASTICSEARCH=ON
@@ -252,7 +250,6 @@ jobs:
-DENABLE_BUSTACHE=ON
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
- -DENABLE_COAP=OFF
-DENABLE_CONTROLLER=ON
-DENABLE_ELASTICSEARCH=OFF
-DENABLE_ENCRYPT_CONFIG=ON
@@ -357,7 +354,6 @@ jobs:
-DENABLE_BUSTACHE=ON
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
- -DENABLE_COAP=ON
-DENABLE_CONTROLLER=ON
-DENABLE_ELASTICSEARCH=ON
-DENABLE_ENCRYPT_CONFIG=ON
@@ -527,7 +523,7 @@ jobs:
run: |
# rocky build can run out of the github runners' disk space if built with RelWithDebInfo so we keep the Release build here
mkdir build && cd build && cmake -DUSE_SHARED_LIBS=ON -DCI_BUILD=ON -DCMAKE_BUILD_TYPE=Release -DSTRICT_GSL_CHECKS=AUDIT -DMINIFI_FAIL_ON_WARNINGS=OFF -DENABLE_AWS=ON -DENABLE_AZURE=ON \
- -DENABLE_COAP=ON -DENABLE_ENCRYPT_CONFIG=ON -DENABLE_GPS=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_NANOFI=ON -DENABLE_OPC=ON \
+ -DENABLE_ENCRYPT_CONFIG=ON -DENABLE_GPS=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_NANOFI=ON -DENABLE_OPC=ON \
-DENABLE_OPENCV=ON -DENABLE_OPENWSMAN=ON -DENABLE_OPS=ON -DENABLE_SENSORS=ON -DENABLE_SQL=ON -DENABLE_SYSTEMD=ON \
-DENABLE_USB_CAMERA=ON -DENABLE_PYTHON_SCRIPTING=ON -DENABLE_LUA_SCRIPTING=ON -DENABLE_KUBERNETES=ON -DENABLE_GCP=ON -DENABLE_PROCFS=ON -DENABLE_PROMETHEUS=ON \
-DENABLE_ELASTICSEARCH=ON -DENABLE_GRAFANA_LOKI=ON -DDOCKER_SKIP_TESTS=OFF -DDOCKER_BUILD_ONLY=ON -DDOCKER_CCACHE_DUMP_LOCATION=${{ env.CCACHE_DIR }} .. && make rocky-test
diff --git a/C2.md b/C2.md
index 27165a41b..9d84974ff 100644
--- a/C2.md
+++ b/C2.md
@@ -27,15 +27,14 @@ options defined are located in minifi.properties.
- [Configuration](#configuration)
- [Base Options](#base-options)
- [Metrics](#metrics)
- - [Protocols](#protocols)
- [UpdatePolicies](#updatepolicies)
- [Triggers](#triggers)
- [C2 File triggers](#c2-file-triggers)
## Description
-Apache NiFi MiNiFi C++ can communicates with a C2 Server via a number of protocols. These protocols
-deliver a C2 response the server, expecting requests in a heartbeat response. The protocol transforms
+Apache NiFi MiNiFi C++ can communicates with a C2 Server via HTTP REST protocol. The protocol
+delivers a C2 response the server, expecting requests in a heartbeat response. The protocol transforms
the C2 messages into a protocol specific representation. The internal representation is an AST therefore
you must define the root classes, which configure the classes that branch from the root. You can define
arbitrary nodes and sub-trees, but this isn't necessary and only advantageous for custom C2 servers. That
@@ -71,13 +70,6 @@ be requested via C2 DESCRIBE manifest command.
# the default is
nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
- # specify C2 protocol -- default is RESTSender if not specified
- nifi.c2.agent.protocol.class=RESTSender
- # nifi.c2.agent.protocol.class=CoapProtocol
-
- # Coap protocol can be also defined in the flow configuration
- # nifi.c2.coap.connector.service=MyCoapProtocol
-
# control c2 heartbeat interval
nifi.c2.agent.heartbeat.period=30 sec
@@ -236,30 +228,6 @@ configuration produces the following JSON:
}
}
-### Protocols
-
-The default protocol is a RESTFul service. The CoAP protocol requires that COAP be enabled either
-through the bootstrap or the cmake flag -DENABLE_COAP=ON .
-
-Once configured, COAP uses a controller service within the flow OR minifi properties entries: nifi.c2.agent.coap.host and nifi.c2.agent.coap.port.
-Note that with CoAP, the payload will be significantly smaller, paring down metrics that are sent in each heartbeat. This will be useful for
-constrained environments.
-
- nifi.c2.agent.coap.host=hostname
- nifi.c2.agent.coap.port=<port number>
-
-If you wish to use the Controller service you made add a controller service named CoapConnectorService with the properties in the example config
-below. Note that Max Queue Size is the only non-required property:
-
- Controller Services:
- - id: 94491a38-015a-1000-0000-000000000001
- name: coapservice
- class: CoapConnectorService
- Properties:
- Remote Server: server
- Remote Port: port
- Max Queue Size: 1000
-
### UpdatePolicies
Updates to MiNiFi C++ properties can be controlled through an UpdatePolicyControllerService named
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 7d258fcf3..7656759a4 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -746,7 +746,7 @@ if (MINIFI_ADVANCED_CODE_COVERAGE)
setup_target_for_coverage_gcovr_html(
NAME coverage
EXCLUDE "build/*" "cmake/*" "minifi_main/*" "nanofi/*" "thirdparty/*" "libminifi/test/*" "encrypt-config/tests/*" "extensions/aws/tests/*"
- "extensions/civetweb/tests/*" "extensions/coap/tests/*" "extensions/elasticsearch/tests/*" "extensions/expression-language/tests/*" "extensions/gcp/tests/*" "extensions/grafana-loki/tests/*"
+ "extensions/civetweb/tests/*" "extensions/elasticsearch/tests/*" "extensions/expression-language/tests/*" "extensions/gcp/tests/*" "extensions/grafana-loki/tests/*"
"extensions/kubernetes/tests/*" "extensions/librdkafka/tests/*" "extensions/lua/tests/*" "extensions/mqtt/tests/*" "extensions/opencv/tests/*" "extensions/procfs/tests/*" "extensions/prometheus/tests/*"
"extensions/script/tests/*" "extensions/sftp/tests/*" "extensions/splunk/tests/*" "extensions/standard-processors/tests/*" "extensions/systemd/tests/*" "extensions/test-processors/*"
"controller/MiNiFiController.cpp"
diff --git a/LICENSE b/LICENSE
index d6d1409b3..f74e10119 100644
--- a/LICENSE
+++ b/LICENSE
@@ -1610,35 +1610,6 @@ OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.
-This product bundles 'libcoap' which is available under a 2-clause BSD license:
-
-Copyright (c) 2010--2019, Olaf Bergmann and others
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are
-met:
-
- o Redistributions of source code must retain the above copyright
- notice, this list of conditions and the following disclaimer.
-
- o Redistributions in binary form must reproduce the above copyright
- notice, this list of conditions and the following disclaimer in
- the documentation and/or other materials provided with the
- distribution.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
-LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
-A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
-HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
-SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
-LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
-DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
-THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
========================================================================
getopt.c
diff --git a/README.md b/README.md
index d078f7ef0..bf268fbb5 100644
--- a/README.md
+++ b/README.md
@@ -355,7 +355,6 @@ This will set up a virtual environment in the bootstrap folder, and guide you th
K. Bustache Support ............Disabled
L. Lua Scripting Support .......Enabled
M. MQTT Support ................Enabled
- N. COAP Support ................Disabled
O. SFTP Support ................Disabled
S. AWS Support .................Enabled
T. OpenCV Support ..............Disabled
diff --git a/Windows.md b/Windows.md
index bea1523df..a7097b32a 100644
--- a/Windows.md
+++ b/Windows.md
@@ -78,7 +78,6 @@ After the build directory it will take optional parameters modifying the CMake c
| /NO_KAFKA | Disables Kafka extension |
| /J | Enables JNI |
| /NO_SQL | Disables SQL extension |
-| /C | Enables CoAP extension |
| /NO_AWS | Disables AWS extension |
| /SFTP | Enables SFTP extension |
| /PDH | Enables Performance Monitor extension |
@@ -135,7 +134,7 @@ A basic working CMake configuration can be inferred from the `win_build_vs.bat`.
```
mkdir build
cd build
-cmake -G "Visual Studio 17 2022" -A x64 -DMINIFI_INCLUDE_VC_REDIST_MERGE_MODULES=OFF -DTEST_CUSTOM_WEL_PROVIDER=OFF -DENABLE_SQL=OFF -DMINIFI_USE_REAL_ODBC_TEST_DRIVER=OFF -DCMAKE_BUILD_TYPE_INIT=Release -DCMAKE_BUILD_TYPE=Release -DWIN32=WIN32 -DENABLE_LIBRDKAFKA=OFF -DENABLE_JNI=OFF -DENABLE_COAP=OFF -DENABLE_AWS=OFF -DENABLE_PDH= -DENABLE_AZURE=OFF -DENABLE_SFTP=OFF -DENABLE_SPLUNK= -DENABLE_GCP= -DENABLE_NANOFI=OFF -DENABLE_OPENCV=OFF -DENABLE_PROMETHEUS=OFF -DENABLE_ELASTICSEARCH= - [...]
+cmake -G "Visual Studio 17 2022" -A x64 -DMINIFI_INCLUDE_VC_REDIST_MERGE_MODULES=OFF -DTEST_CUSTOM_WEL_PROVIDER=OFF -DENABLE_SQL=OFF -DMINIFI_USE_REAL_ODBC_TEST_DRIVER=OFF -DCMAKE_BUILD_TYPE_INIT=Release -DCMAKE_BUILD_TYPE=Release -DWIN32=WIN32 -DENABLE_LIBRDKAFKA=OFF -DENABLE_JNI=OFF -DENABLE_AWS=OFF -DENABLE_PDH= -DENABLE_AZURE=OFF -DENABLE_SFTP=OFF -DENABLE_SPLUNK= -DENABLE_GCP= -DENABLE_NANOFI=OFF -DENABLE_OPENCV=OFF -DENABLE_PROMETHEUS=OFF -DENABLE_ELASTICSEARCH= -DUSE_SHARED_LIBS=O [...]
msbuild /m nifi-minifi-cpp.sln /property:Configuration=Release /property:Platform=x64
copy minifi_main\Release\minifi.exe minifi_main\
cpack
diff --git a/bootstrap.sh b/bootstrap.sh
index a376be527..8e3513720 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -291,11 +291,6 @@ add_option KUBERNETES_ENABLED ${TRUE} "ENABLE_KUBERNETES"
add_option MQTT_ENABLED ${TRUE} "ENABLE_MQTT"
-add_option COAP_ENABLED ${FALSE} "ENABLE_COAP"
-add_dependency COAP_ENABLED "automake"
-add_dependency COAP_ENABLED "autoconf"
-add_dependency COAP_ENABLED "libtool"
-
add_option JNI_ENABLED ${FALSE} "ENABLE_JNI"
add_dependency JNI_ENABLED "jnibuild"
diff --git a/bootstrap/system_dependency.py b/bootstrap/system_dependency.py
index 3ef31dc70..8abff4183 100644
--- a/bootstrap/system_dependency.py
+++ b/bootstrap/system_dependency.py
@@ -37,7 +37,7 @@ def _create_system_dependencies(minifi_options: MinifiOptions) -> Dict[str, Set[
system_dependencies['libpng'] = {'libpng'}
if minifi_options.is_enabled("ENABLE_GPS"):
system_dependencies['gpsd'] = {'gpsd'}
- if minifi_options.is_enabled("ENABLE_COAP") or minifi_options.is_enabled("ENABLE_SQL"):
+ if minifi_options.is_enabled("ENABLE_SQL"):
system_dependencies['automake'] = {'automake'}
system_dependencies['autoconf'] = {'autoconf'}
system_dependencies['libtool'] = {'libtool'}
diff --git a/bstrp_functions.sh b/bstrp_functions.sh
index 03d4c8a1e..8eb5fd1e9 100755
--- a/bstrp_functions.sh
+++ b/bstrp_functions.sh
@@ -370,7 +370,6 @@ show_supported_features() {
echo "K. Bustache Support ............$(print_feature_status BUSTACHE_ENABLED)"
echo "L. Lua Scripting Support .......$(print_feature_status LUA_SCRIPTING_ENABLED)"
echo "M. MQTT Support ................$(print_feature_status MQTT_ENABLED)"
- echo "N. COAP Support ................$(print_feature_status COAP_ENABLED)"
echo "O. SFTP Support ................$(print_feature_status SFTP_ENABLED)"
echo "S. AWS Support .................$(print_feature_status AWS_ENABLED)"
echo "T. OpenCV Support ..............$(print_feature_status OPENCV_ENABLED)"
@@ -426,7 +425,6 @@ read_feature_options(){
k) ToggleFeature BUSTACHE_ENABLED ;;
l) ToggleFeature LUA_SCRIPTING_ENABLED ;;
m) ToggleFeature MQTT_ENABLED ;;
- n) ToggleFeature COAP_ENABLED ;;
o) ToggleFeature SFTP_ENABLED ;;
s) ToggleFeature AWS_ENABLED ;;
t) ToggleFeature OPENCV_ENABLED ;;
diff --git a/cmake/BundledLibCoAP.cmake b/cmake/BundledLibCoAP.cmake
deleted file mode 100644
index 8c4703390..000000000
--- a/cmake/BundledLibCoAP.cmake
+++ /dev/null
@@ -1,85 +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.
-#
-
-function(use_bundled_libcoap SOURCE_DIR BINARY_DIR)
- message("Using bundled libcoap")
-
- # Define patch step
- if (WIN32)
- set(PC "${Patch_EXECUTABLE}" -p1 -i "${SOURCE_DIR}/thirdparty/libcoap/libcoap-windows-cmake.patch")
- endif()
-
- # Define byproducts
- if (WIN32)
- set(BYPRODUCT "lib/coap.lib")
- else()
- set(BYPRODUCT "lib/libcoap-2.a")
- endif()
-
- # Build project
- set(LIBCOAP_URL https://github.com/obgm/libcoap/archive/v4.2.1.tar.gz)
- set(LIBCOAP_URL_HASH "SHA256=29a0394a265d3febee41e5e2dc03d34292a0aede37f5f80334e529ac0dab2321")
-
- if (WIN32)
- set(LIBCOAP_CMAKE_ARGS ${PASSTHROUGH_CMAKE_ARGS}
- "-DCMAKE_INSTALL_PREFIX=${BINARY_DIR}/thirdparty/libcoap-install")
-
- ExternalProject_Add(
- coap-external
- URL ${LIBCOAP_URL}
- URL_HASH ${LIBCOAP_URL_HASH}
- CMAKE_ARGS ${LIBCOAP_CMAKE_ARGS}
- PATCH_COMMAND ${PC}
- BUILD_BYPRODUCTS "${BINARY_DIR}/thirdparty/libcoap-install/${BYPRODUCT}"
- EXCLUDE_FROM_ALL TRUE
- )
- else()
- ExternalProject_Add(
- coap-external
- URL ${LIBCOAP_URL}
- URL_HASH ${LIBCOAP_URL_HASH}
- BUILD_IN_SOURCE true
- SOURCE_DIR "${BINARY_DIR}/thirdparty/libcoap-src"
- BUILD_COMMAND make
- CMAKE_COMMAND ""
- UPDATE_COMMAND ""
- INSTALL_COMMAND make install
- BUILD_BYPRODUCTS "${BINARY_DIR}/thirdparty/libcoap-install/${BYPRODUCT}"
- CONFIGURE_COMMAND ""
- PATCH_COMMAND ./autogen.sh && ./configure "CFLAGS=${PASSTHROUGH_CMAKE_C_FLAGS} -fPIC" "CXXFLAGS=${PASSTHROUGH_CMAKE_CXX_FLAGS} -fPIC" --with-pic --disable-examples --disable-dtls --disable-tests --disable-documentation --prefix=${BINARY_DIR}/thirdparty/libcoap-install
- STEP_TARGETS build
- EXCLUDE_FROM_ALL TRUE
- )
- endif()
-
- # Set variables
- set(COAP_FOUND "YES" CACHE STRING "" FORCE)
- set(COAP_INCLUDE_DIRS "${BINARY_DIR}/thirdparty/libcoap-install/include" CACHE STRING "" FORCE)
- set(COAP_LIBRARY "${BINARY_DIR}/thirdparty/libcoap-install/${BYPRODUCT}" CACHE STRING "" FORCE)
- set(COAP_LIBRARIES "${COAP_LIBRARY}" CACHE STRING "" FORCE)
-
- # Create imported targets
- file(MAKE_DIRECTORY ${COAP_INCLUDE_DIRS})
-
- add_library(COAP::libcoap STATIC IMPORTED)
- set_target_properties(COAP::libcoap PROPERTIES IMPORTED_LOCATION "${COAP_LIBRARY}")
- add_dependencies(COAP::libcoap coap-external)
- set_property(TARGET COAP::libcoap APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES "${COAP_INCLUDE_DIRS}")
- set_property(TARGET COAP::libcoap APPEND PROPERTY INTERFACE_LINK_LIBRARIES Threads::Threads)
-endfunction(use_bundled_libcoap)
diff --git a/cmake/DockerConfig.cmake b/cmake/DockerConfig.cmake
index e0c902198..f343ffb8a 100644
--- a/cmake/DockerConfig.cmake
+++ b/cmake/DockerConfig.cmake
@@ -65,7 +65,6 @@ add_custom_target(
-DENABLE_GRPC_FOR_LOKI=OFF
-DENABLE_PROCFS=OFF
-DENABLE_GPS=OFF
- -DENABLE_COAP=OFF
-DENABLE_PCAP=OFF
-DENABLE_SENSORS=OFF
-DENABLE_USB_CAMERA=OFF
diff --git a/cmake/MiNiFiOptions.cmake b/cmake/MiNiFiOptions.cmake
index b6f8a7ae4..40f33aa5a 100644
--- a/cmake/MiNiFiOptions.cmake
+++ b/cmake/MiNiFiOptions.cmake
@@ -97,7 +97,6 @@ add_minifi_option(ENABLE_LIBARCHIVE "Enables the lib archive extensions." ON)
add_minifi_option(ENABLE_LZMA "Enables the liblzma build" ON)
add_minifi_option(ENABLE_BZIP2 "Enables the bzip2 build" ON)
add_minifi_option(ENABLE_GPS "Enables the GPS extension." OFF)
-add_minifi_option(ENABLE_COAP "Enables the CoAP extension." OFF)
add_minifi_option(ENABLE_SQL "Enables the SQL Suite of Tools." ON)
add_minifi_option(ENABLE_MQTT "Enables the mqtt extension." ON)
add_minifi_option(ENABLE_PCAP "Enables the PCAP extension." OFF)
diff --git a/conf/minifi.properties b/conf/minifi.properties
index e24f3706e..477ebfd35 100644
--- a/conf/minifi.properties
+++ b/conf/minifi.properties
@@ -81,13 +81,6 @@ nifi.content.repository.class.name=DatabaseContentRepository
## Enabling C2 Uncomment each of the following options
## define those with missing options
#nifi.c2.enable=true
-## define protocol parameters
-## The default is RESTSender.
-## Alternatively, you may use CoapProtocol if that extension is built.
-#nifi.c2.agent.protocol.class=RESTSender
-#nifi.c2.agent.protocol.class=CoapProtocol
-#nifi.c2.agent.coap.host=
-#nifi.c2.agent.coap.port=
## base URL of the c2 server,
## very likely the same base url of rest urls
#nifi.c2.flow.base.url=
diff --git a/encrypt-config/tests/ConfigFileEncryptorTests.cpp b/encrypt-config/tests/ConfigFileEncryptorTests.cpp
index 4fb309c06..d489c175f 100644
--- a/encrypt-config/tests/ConfigFileEncryptorTests.cpp
+++ b/encrypt-config/tests/ConfigFileEncryptorTests.cpp
@@ -77,7 +77,7 @@ TEST_CASE("ConfigFileEncryptor can encrypt the sensitive properties", "[encrypt-
uint32_t num_properties_encrypted = encryptSensitivePropertiesInFile(test_file, KEY);
REQUIRE(num_properties_encrypted == 1);
- REQUIRE(test_file.size() == 115);
+ REQUIRE(test_file.size() == 108);
REQUIRE(check_encryption(test_file, Configuration::nifi_rest_api_password, original_password.length()));
SECTION("calling encryptSensitiveValuesInMinifiProperties a second time does nothing") {
diff --git a/encrypt-config/tests/ConfigFileTests.cpp b/encrypt-config/tests/ConfigFileTests.cpp
index 0c72f1a53..252d1aaac 100644
--- a/encrypt-config/tests/ConfigFileTests.cpp
+++ b/encrypt-config/tests/ConfigFileTests.cpp
@@ -90,7 +90,7 @@ TEST_CASE("ConfigFile creates an empty object from a nonexistent file", "[encryp
TEST_CASE("ConfigFile can parse a simple config file", "[encrypt-config][constructor]") {
ConfigFile test_file{std::ifstream{"resources/minifi.properties"}};
- REQUIRE(test_file.size() == 114);
+ REQUIRE(test_file.size() == 107);
}
TEST_CASE("ConfigFile can test whether a key is present", "[encrypt-config][hasValue]") {
@@ -102,7 +102,7 @@ TEST_CASE("ConfigFile can test whether a key is present", "[encrypt-config][hasV
TEST_CASE("ConfigFile can read empty properties correctly", "[encrypt-config][constructor]") {
ConfigFile test_file{std::ifstream{"resources/with-additional-sensitive-props.minifi.properties"}};
- REQUIRE(test_file.size() == 115);
+ REQUIRE(test_file.size() == 108);
auto empty_property = test_file.getValue(Configuration::nifi_security_need_ClientAuth);
REQUIRE(empty_property);
@@ -143,7 +143,7 @@ TEST_CASE("ConfigFile can add a new setting after an existing setting", "[encryp
SECTION("valid key") {
test_file.insertAfter(Configuration::nifi_rest_api_password, "nifi.rest.api.password.protected", "my-cipher-name");
- REQUIRE(test_file.size() == 115);
+ REQUIRE(test_file.size() == 108);
REQUIRE(test_file.getValue("nifi.rest.api.password.protected") == "my-cipher-name");
}
@@ -158,7 +158,7 @@ TEST_CASE("ConfigFile can add a new setting at the end", "[encrypt-config][appen
const std::string KEY = "nifi.bootstrap.sensitive.key";
const std::string VALUE = "aa411f289c91685ef9d5a9e5a4fad9393ff4c7a78ab978484323488caed7a9ab";
test_file.append(KEY, VALUE);
- REQUIRE(test_file.size() == 115);
+ REQUIRE(test_file.size() == 108);
REQUIRE(test_file.getValue(KEY) == std::make_optional(VALUE));
}
diff --git a/encrypt-config/tests/resources/minifi.properties b/encrypt-config/tests/resources/minifi.properties
index 7df24ba7f..2f0cd2f08 100644
--- a/encrypt-config/tests/resources/minifi.properties
+++ b/encrypt-config/tests/resources/minifi.properties
@@ -50,13 +50,6 @@ nifi.rest.api.password=password
## Enabling C2 Uncomment each of the following options
## define those with missing options
nifi.c2.enable=true
-## define protocol parameters
-## The default is RESTSender.
-## Alternatively, you may use CoapProtocol if that extension is built.
-nifi.c2.agent.protocol.class=RESTSender
-#nifi.c2.agent.protocol.class=CoapProtocol
-#nifi.c2.agent.coap.host=
-#nifi.c2.agent.coap.port=
## base URL of the c2 server,
## very likely the same base url of rest urls
nifi.c2.flow.base.url=http://localhost:10080/c2-server/api
diff --git a/encrypt-config/tests/resources/with-additional-sensitive-props.minifi.properties b/encrypt-config/tests/resources/with-additional-sensitive-props.minifi.properties
index 60e6ff61c..655198b2a 100644
--- a/encrypt-config/tests/resources/with-additional-sensitive-props.minifi.properties
+++ b/encrypt-config/tests/resources/with-additional-sensitive-props.minifi.properties
@@ -52,13 +52,6 @@ nifi.rest.api.password=password
## Enabling C2 Uncomment each of the following options
## define those with missing options
nifi.c2.enable=true
-## define protocol parameters
-## The default is RESTSender.
-## Alternatively, you may use CoapProtocol if that extension is built.
-nifi.c2.agent.protocol.class=RESTSender
-#nifi.c2.agent.protocol.class=CoapProtocol
-#nifi.c2.agent.coap.host=
-#nifi.c2.agent.coap.port=
## base URL of the c2 server,
## very likely the same base url of rest urls
nifi.c2.flow.base.url=http://localhost:10080/c2-server/api
diff --git a/extensions/coap/CMakeLists.txt b/extensions/coap/CMakeLists.txt
deleted file mode 100644
index 3a374335a..000000000
--- a/extensions/coap/CMakeLists.txt
+++ /dev/null
@@ -1,42 +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.
-#
-
-if (NOT (ENABLE_ALL OR ENABLE_COAP STREQUAL "ON"))
- return()
-endif()
-
-include(BundledLibCoAP)
-use_bundled_libcoap(${CMAKE_SOURCE_DIR} ${CMAKE_BINARY_DIR})
-
-include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
-include_directories(protocols nanofi controllerservice server)
-include_directories(../http-curl/)
-
-file(GLOB CSOURCES "nanofi/*.c")
-file(GLOB SOURCES "*.cpp" "protocols/*.cpp" "processors/*.cpp" "controllerservice/*.cpp" "server/*.cpp" )
-
-add_minifi_library(nanofi-coap-c STATIC ${CSOURCES})
-add_minifi_library(minifi-coap SHARED ${SOURCES})
-
-target_link_libraries(nanofi-coap-c COAP::libcoap Threads::Threads)
-target_link_libraries(minifi-coap ${LIBMINIFI})
-target_link_libraries(minifi-coap nanofi-coap-c Threads::Threads COAP::libcoap)
-
-register_extension(minifi-coap "COAP EXTENSIONS" COAP-EXTENSION "Enables LibCOAP Functionality." "extensions/coap/tests/")
-register_extension_linter(minifi-coap-extensions-linter)
diff --git a/extensions/coap/COAPLoader.cpp b/extensions/coap/COAPLoader.cpp
deleted file mode 100644
index 5fb21c2f0..000000000
--- a/extensions/coap/COAPLoader.cpp
+++ /dev/null
@@ -1,47 +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.
- */
-
-#include "core/logging/LoggerConfiguration.h"
-#include "core/extension/Extension.h"
-
-#ifdef WIN32
-#include <winsock2.h>
-#endif
-
-static bool init(const std::shared_ptr<org::apache::nifi::minifi::Configure> & /*config*/) {
-#ifdef WIN32
- static WSADATA s_wsaData;
- int iWinSockInitResult = WSAStartup(MAKEWORD(2, 2), &s_wsaData);
- if (iWinSockInitResult != 0) {
- core::logging::LoggerFactory<COAPObjectFactoryInitializer>::getLogger()->log_error("WSAStartup failed with error {}", iWinSockInitResult);
- return false;
- } else {
- return true;
- }
-#else
- return true;
-#endif
-}
-
-static void deinit() {
-#ifdef WIN32
- WSACleanup();
-#endif
-}
-
-REGISTER_EXTENSION("CoapExtension", init, deinit);
diff --git a/extensions/coap/controllerservice/CoapConnector.cpp b/extensions/coap/controllerservice/CoapConnector.cpp
deleted file mode 100644
index 4b2b23da9..000000000
--- a/extensions/coap/controllerservice/CoapConnector.cpp
+++ /dev/null
@@ -1,74 +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.
- */
-
-#include "CoapConnector.h"
-
-#include <string>
-#include <memory>
-
-#include "core/logging/LoggerConfiguration.h"
-#include "core/controller/ControllerService.h"
-#include "core/Resource.h"
-#include "io/validation.h"
-#include "properties/Configure.h"
-
-namespace org::apache::nifi::minifi::coap::controllers {
-
-void CoapConnectorService::initialize() {
- std::lock_guard<std::mutex> lock(initialization_mutex_);
- if (initialized_) {
- return;
- }
-
- CoapMessaging::getInstance();
-
- ControllerService::initialize();
-
- initializeProperties();
-
- initialized_ = true;
-}
-
-void CoapConnectorService::onEnable() {
- std::string port_str;
- if (getProperty(RemoteServer, host_) && !host_.empty() && getProperty(Port, port_str) && !port_str.empty()) {
- core::Property::StringToInt(port_str, port_);
- } else {
- // this is the case where we aren't being used in the context of a single controller service.
- if (configuration_->get(Configuration::nifi_c2_agent_coap_host, host_) && configuration_->get(Configuration::nifi_c2_agent_coap_port, port_str)) {
- core::Property::StringToInt(port_str, port_);
- }
- }
-}
-
-CoapResponse CoapConnectorService::sendPayload(uint8_t type, const std::string &endpoint, const CoapMessage *message) {
- // internally we are dealing with CoAPMessage in the two way communication, but the C++ ControllerService
- // will provide a CoAPResponse
- auto pdu = create_connection(type, host_.c_str(), endpoint.c_str(), port_, message);
- send_pdu(pdu);
- auto response = CoapMessaging::getInstance().pop(pdu->ctx);
- free_pdu(pdu);
- return response;
-}
-
-void CoapConnectorService::initializeProperties() {
- setSupportedProperties(Properties);
-}
-
-REGISTER_RESOURCE(CoapConnectorService, InternalResource);
-
-} // namespace org::apache::nifi::minifi::coap::controllers
diff --git a/extensions/coap/controllerservice/CoapConnector.h b/extensions/coap/controllerservice/CoapConnector.h
deleted file mode 100644
index b328a6d28..000000000
--- a/extensions/coap/controllerservice/CoapConnector.h
+++ /dev/null
@@ -1,126 +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.
- */
-#pragma once
-
-#include <memory>
-#include <string>
-#include <mutex>
-#include <atomic>
-#include <utility>
-
-#include "core/logging/LoggerConfiguration.h"
-#include "core/controller/ControllerService.h"
-#include "core/PropertyDefinition.h"
-#include "core/PropertyDefinitionBuilder.h"
-#include "core/PropertyType.h"
-
-#include "CoapResponse.h"
-#include "CoapMessaging.h"
-#include "coap_functions.h"
-#include "coap_connection.h"
-#include "coap_message.h"
-
-namespace org::apache::nifi::minifi::coap::controllers {
-
-/**
- * Purpose and Justification: Controller services function as a layerable way to provide
- * services to internal services. While a controller service is generally configured from the flow,
- * we want to follow the open closed principle and provide CoAP services to other components.
- */
-class CoapConnectorService : public core::controller::ControllerService {
- public:
- explicit CoapConnectorService(std::string name, const utils::Identifier &uuid = {})
- : ControllerService(std::move(name), uuid) {
- initialize();
- }
-
- explicit CoapConnectorService(const std::string &name, const std::shared_ptr<Configure> &configuration)
- : ControllerService(name) {
- setConfiguration(configuration);
- initialize();
- }
-
- EXTENSIONAPI static constexpr auto RemoteServer = core::PropertyDefinitionBuilder<>::createProperty("Remote Server")
- .withDescription("Remote CoAP server")
- .isRequired(false)
- .build();
- EXTENSIONAPI static constexpr auto Port = core::PropertyDefinitionBuilder<>::createProperty("Remote Port")
- .withDescription("Remote CoAP server port")
- .withPropertyType(core::StandardPropertyTypes::UNSIGNED_LONG_TYPE)
- .withDefaultValue("8181")
- .isRequired(true)
- .build();
- EXTENSIONAPI static constexpr auto MaxQueueSize = core::PropertyDefinitionBuilder<>::createProperty("Max Queue Size")
- .withDescription("Max queue size for received data ")
- .withPropertyType(core::StandardPropertyTypes::UNSIGNED_LONG_TYPE)
- .withDefaultValue("1000")
- .isRequired(false)
- .build();
- EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 3>{
- RemoteServer,
- Port,
- MaxQueueSize
- };
-
- EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_CONTROLLER_SERVICES
-
- void initialize() override;
-
- void yield() override { }
-
- bool isRunning() const override {
- return getState() == core::controller::ControllerServiceState::ENABLED;
- }
-
- bool isWorkAvailable() override {
- return false;
- }
-
- void onEnable() override;
-
- /**
- * Sends the payload to the endpoint, returning the response as we await. Will retry transmission
- * @param type type of payload to endpoint interaction ( GET, POST, PUT, DELETE ).
- * @param end endpoint is the connecting endpoint on the server
- * @param payload is the data to be sent
- * @param size size of the payload to be sent
- * @return CoAPMessage that contains the response code and data, if any.
- */
- CoapResponse sendPayload(uint8_t type, const std::string &endpoint, const CoapMessage * message);
-
- protected:
- void initializeProperties();
-
- // connector mutex to controll access to the mapping, above.
- std::mutex connector_mutex_;
-
- // initialization mutex.
- std::mutex initialization_mutex_;
-
- bool initialized_ = false;
-
- private:
- // host connecting to.
- std::string host_;
- // port connecting to
- unsigned int port_{ 0 };
-
- std::shared_ptr<core::logging::Logger> logger_{ core::logging::LoggerFactory<CoapConnectorService>::getLogger() };
-};
-
-} // namespace org::apache::nifi::minifi::coap::controllers
diff --git a/extensions/coap/controllerservice/CoapMessaging.h b/extensions/coap/controllerservice/CoapMessaging.h
deleted file mode 100644
index e75f41dae..000000000
--- a/extensions/coap/controllerservice/CoapMessaging.h
+++ /dev/null
@@ -1,109 +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.
- */
-#pragma once
-
-#include <unordered_map>
-#include <mutex>
-#include <utility>
-
-#include "CoapResponse.h"
-#include "coap_functions.h"
-#include "coap_connection.h"
-#include "coap_message.h"
-
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace coap {
-namespace controllers {
-
-class CoapMessaging {
- public:
- static CoapMessaging &getInstance() {
- static CoapMessaging instance;
- return instance;
- }
-
- /**
- * Determines if the pointer is present in the internal map.
- */
- bool hasResponse(coap_context_t *ctx) const {
- std::lock_guard<std::mutex> lock(connector_mutex_);
- return messages_.find(ctx) != messages_.end();
- }
- /**
- * Returns a response if one exists.
- */
- CoapResponse pop(const coap_context_t * const ctx) {
- CoapResponse response(-1);
- std::lock_guard<std::mutex> lock(connector_mutex_);
- auto msg = messages_.find(const_cast<coap_context_t*>(ctx));
- if (msg != std::end(messages_)) {
- response = std::move(msg->second);
- messages_.erase(const_cast<coap_context_t*>(ctx));
- }
- return response;
- }
-
- protected:
- /**
- * Intended to receive errors from the context.
- */
- static void receiveError(void *receiver_context, coap_context_t *ctx, unsigned int code) {
- CoapMessaging *connector = static_cast<CoapMessaging*>(receiver_context);
- CoapResponse message(code);
- connector->enqueueResponse(ctx, std::move(message));
- }
-
- /**
- * Receives messages from the context.
- */
- static void receiveMessage(void *receiver_context, coap_context_t *ctx, CoapMessage * const msg) {
- CoapMessaging *connector = static_cast<CoapMessaging*>(receiver_context);
- CoapResponse message(msg);
- connector->enqueueResponse(ctx, std::move(message));
- }
-
- void enqueueResponse(coap_context_t *ctx, CoapResponse &&msg) {
- std::lock_guard<std::mutex> lock(connector_mutex_);
- messages_.insert(std::make_pair(ctx, std::move(msg)));
- }
-
- private:
- /**
- * Private constructor since this is intended to be a singleton
- */
- CoapMessaging() {
- callback_pointers ptrs;
- ptrs.data_received = receiveMessage;
- ptrs.received_error = receiveError;
- init_coap_api(this, &ptrs);
- }
- // connector mutex. mutable since it's used within hasResponse.
- mutable std::mutex connector_mutex_;
- // map of messages based on the context. We only allow a single message per context
- // at any given time.
- std::unordered_map<coap_context_t*, CoapResponse> messages_;
-};
-} /* namespace controllers */
-} /* namespace coap */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
diff --git a/extensions/coap/controllerservice/CoapResponse.h b/extensions/coap/controllerservice/CoapResponse.h
deleted file mode 100644
index f7d3f0ba0..000000000
--- a/extensions/coap/controllerservice/CoapResponse.h
+++ /dev/null
@@ -1,82 +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.
- */
-#pragma once
-
-#include <cstddef>
-#include <memory>
-#include <vector>
-
-#include "coap_message.h"
-
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace coap {
-namespace controllers {
-/**
- * Purpose and Justification :CoapMessage is in internal message format that is sent to and from consumers of this controller service.
- *
- */
-class CoapResponse {
- public:
- /**
- * Creates a CoAPResponse to a CoAPMessage. Takes ownership of the argument
- * and copies the data.
- */
- explicit CoapResponse(CoapMessage* const msg)
- : code_(msg->code_) {
- data_.resize(msg->size_);
- memcpy(data_.data(), msg->data_, data_.size());
- free_coap_message(msg);
- }
-
- explicit CoapResponse(uint32_t code)
- : code_(code) {
- }
-
- CoapResponse(const CoapResponse &other) = delete;
- CoapResponse(CoapResponse &&other) = default;
- ~CoapResponse() = default;
-
- [[nodiscard]] std::span<const std::byte> getData() const noexcept {
- return data_;
- }
-
- /**
- * Returns the response code from the data.
- * @return data.
- */
- [[nodiscard]] uint32_t getCode() const {
- return code_;
- }
-
- CoapResponse &operator=(const CoapResponse &other) = delete;
- CoapResponse &operator=(CoapResponse &&other) = default;
-
- private:
- uint32_t code_;
- std::vector<std::byte> data_;
-};
-
-} /* namespace controllers */
-} /* namespace coap */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
diff --git a/extensions/coap/nanofi/coap_connection.c b/extensions/coap/nanofi/coap_connection.c
deleted file mode 100644
index f44a2258e..000000000
--- a/extensions/coap/nanofi/coap_connection.c
+++ /dev/null
@@ -1,124 +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.
- */
-#include "coap_connection.h"
-
-#ifdef WIN32
-#include <winsock2.h>
-#else
-#include <netdb.h>
-#endif
-
-CoapPDU *create_connection(uint8_t type, const char * const server, const char * const endpoint, int port, const CoapMessage * const message) {
- CoapPDU *pdu = (CoapPDU*) malloc(sizeof(CoapPDU));
-
- pdu->ctx = NULL;
- pdu->session = NULL;
-
- coap_uri_t uri;
- uri.host.s = (uint8_t*) server; // may be a loss in resolution, but hostnames should be char *
- uri.host.length = strlen(server);
- uri.path.s = (uint8_t*) endpoint; // ^ same as above for paths.
- uri.path.length = strlen(endpoint);
- uri.port = port;
- uri.scheme = COAP_URI_SCHEME_COAP;
-
- coap_pdu_t* request;
-
- int res = resolve_address(&uri.host, &pdu->dst_addr.addr.sa);
- if (res < 0) {
- return NULL;
- }
- pdu->dst_addr.size = res;
- pdu->dst_addr.addr.sin.sin_port = htons(uri.port);
-
- char port_str[NI_MAXSERV] = "0";
-
- switch (pdu->dst_addr.addr.sa.sa_family) {
- case AF_INET:
- if (!create_session(&pdu->ctx, &pdu->session, 0x00, port_str, &pdu->dst_addr)) {
- break;
- } else {
- return NULL;
- }
- case AF_INET6:
- if (!create_session(&pdu->ctx, &pdu->session, 0x00, port_str, &pdu->dst_addr)) {
- break;
- } else {
- return NULL;
- }
- default:
- ;
- }
-
- // we want to register handlers in the event that an error occurs or nack is returned
- // from the library
- coap_register_event_handler(pdu->ctx, coap_event);
- coap_register_nack_handler(pdu->ctx, no_acknowledgement);
-
- coap_context_set_keepalive(pdu->ctx, 1);
-
- coap_str_const_t pld;
- pld.length = message->size_;
- pld.s = message->data_;
-
- coap_register_option(pdu->ctx, COAP_OPTION_BLOCK2);
-
- // set the response handler
- coap_register_response_handler(pdu->ctx, response_handler);
-
- pdu->session->max_retransmit = 1;
- pdu->optlist = NULL;
-
- // add the URI option to the options list
- coap_insert_optlist(&pdu->optlist, coap_new_optlist(COAP_OPTION_URI_PATH, uri.path.length, uri.path.s));
-
- // next, create the PDU.
- if (!(request = create_request(pdu->ctx, pdu->session, &pdu->optlist, type, &pld)))
- return NULL;
-
- // send the PDU using the session.
- coap_send(pdu->session, request);
- return pdu;
-}
-
-int8_t send_pdu(const CoapPDU * const pdu) {
- uint64_t wait_ms = 1 * 200;
- // run once will attempt to send the first time
- int runResponse = coap_run_once(pdu->ctx, wait_ms);
- // if no data is received, we will attempt re-transmission
- // until the number of attempts has been reached.
- while (!coap_can_exit(pdu->ctx)) {
- runResponse = coap_run_once(pdu->ctx, wait_ms);
- }
- if (runResponse < 0)
- return -1;
- else
- return 0;
-}
-
-int8_t free_pdu(CoapPDU * pdu) {
- if (NULL == pdu) {
- return -1;
- }
- coap_delete_optlist(pdu->optlist);
- coap_session_release(pdu->session);
- coap_free_context(pdu->ctx);
- free(pdu);
- return 0;
-}
-
diff --git a/extensions/coap/nanofi/coap_connection.h b/extensions/coap/nanofi/coap_connection.h
deleted file mode 100644
index 514366eea..000000000
--- a/extensions/coap/nanofi/coap_connection.h
+++ /dev/null
@@ -1,67 +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 EXTENSIONS_COAP_NANOFI_COAP_CONNECTION_H_
-#define EXTENSIONS_COAP_NANOFI_COAP_CONNECTION_H_
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <coap2/coap.h>
-#include "coap_message.h"
-#include "coap_functions.h"
-
-
-
-typedef struct {
- struct coap_context_t* ctx;
- struct coap_session_t* session;
- coap_address_t dst_addr;
- coap_address_t src_addr;
- coap_optlist_t *optlist;
-} CoapPDU;
-
-
-/**
- * Creates a connection to the server
- * @param type connection type
- * @param server server name
- * @param endpoint endpoint to connect to
- * @param port CoAP port
- * @param message message to send
- * @return CoAPPDU object.
- */
-CoapPDU *create_connection(uint8_t type, const char * const server, const char * const endpoint, int port, const CoapMessage * const message);
-
-/**
- * Sends the pdu
- * @param pdu to send
- * @return result 0 if success, failure otherwise
- */
-int8_t send_pdu(const CoapPDU *const pdu);
-/**
- * Frees the pdu
- * @param pdu to free
- * @return result 0 if success, failure otherwise
- */
-
-int8_t free_pdu(CoapPDU *pdu);
-
-#ifdef __cplusplus
-}
-#endif
-#endif /* EXTENSIONS_COAP_NANOFI_COAP_CONNECTION_H_ */
diff --git a/extensions/coap/nanofi/coap_functions.c b/extensions/coap/nanofi/coap_functions.c
deleted file mode 100644
index 3d7fa4273..000000000
--- a/extensions/coap/nanofi/coap_functions.c
+++ /dev/null
@@ -1,231 +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.
- */
-#include "coap_functions.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#ifndef UNUSED
-#define UNUSED(x) (void)(x)
-#endif
-
-// defines the context specific data for the data receiver
-static void *receiver;
-static callback_pointers global_ptrs;
-
-/**
- * Initialize the API access. Not thread safe.
- */
-void init_coap_api(void *rcvr, callback_pointers *ptrs) {
- global_ptrs.data_received = ptrs->data_received;
- global_ptrs.received_error = ptrs->received_error;
- receiver = rcvr;
-}
-
-int create_session(coap_context_t **ctx, coap_session_t **session, const char *node, const char *port, coap_address_t *dst_addr) {
- int getaddrres;
- struct addrinfo hints;
- coap_proto_t proto = COAP_PROTO_UDP;
- struct addrinfo *result;
- struct addrinfo *interface_itr;
-
- memset(&hints, 0, sizeof(struct addrinfo));
- hints.ai_family = AF_UNSPEC; // ipv4 or ipv6
- hints.ai_socktype = COAP_PROTO_RELIABLE(proto) ? SOCK_STREAM : SOCK_DGRAM;
- hints.ai_flags = AI_PASSIVE | AI_NUMERICHOST | AI_NUMERICSERV | AI_ALL;
-
- if (node) {
- getaddrres = getaddrinfo(node, port, &hints, &result);
- if (getaddrres != 0) {
- perror("getaddrinfo");
- return -1;
- }
- for (interface_itr = result; interface_itr != NULL; interface_itr = interface_itr->ai_next) {
- coap_address_t addr;
-
- if (interface_itr->ai_addrlen <= sizeof(addr.addr)) {
- coap_address_init(&addr);
- addr.size = interface_itr->ai_addrlen;
- memcpy(&addr.addr, interface_itr->ai_addr, interface_itr->ai_addrlen);
-
- *ctx = coap_new_context(0x00);
-
- *session = coap_new_client_session(*ctx, &addr, dst_addr, proto);
- if (*ctx && *session) {
- freeaddrinfo(result);
- return 0;
- }
- }
- }
- freeaddrinfo(result);
- return -2;
- } else {
- *ctx = coap_new_context(0x00);
-
- *session = coap_new_client_session(*ctx, 0x00, dst_addr, proto);
- return 0;
- }
-}
-
-int create_endpoint_context(coap_context_t **ctx, const char *node, const char *port) {
- struct addrinfo hints;
- struct addrinfo *result;
- struct addrinfo *interface_itr;
-
- memset(&hints, 0, sizeof(struct addrinfo));
- hints.ai_family = AF_UNSPEC; // ipv4 or ipv6
- hints.ai_socktype = SOCK_DGRAM;
- hints.ai_flags = AI_PASSIVE | AI_NUMERICHOST;
- int getaddrres = getaddrinfo(node, port, &hints, &result);
- if (getaddrres != 0) {
- perror("getaddrinfo");
- return -1;
- }
-
- for (interface_itr = result; interface_itr != NULL; interface_itr = interface_itr->ai_next) {
- coap_address_t addr;
-
- if (interface_itr->ai_addrlen <= sizeof(addr.addr)) {
- coap_address_init(&addr);
- addr.size = interface_itr->ai_addrlen;
- memcpy(&addr.addr, interface_itr->ai_addr, interface_itr->ai_addrlen);
-
- *ctx = coap_new_context(0x00);
-
- coap_endpoint_t * ep_udp = coap_new_endpoint(*ctx, &addr, COAP_PROTO_UDP);
-
- if (*ctx && ep_udp) {
- freeaddrinfo(result);
- return 0;
- }
- }
- }
-
- freeaddrinfo(result);
- return -2;
-}
-
-struct coap_pdu_t *create_request(struct coap_context_t *ctx, struct coap_session_t *session, coap_optlist_t **optlist, unsigned char code, coap_str_const_t *ptr) {
- UNUSED(ctx);
- coap_pdu_t *pdu;
-
- if (!(pdu = coap_new_pdu(session)))
- return NULL;
-
- pdu->type = COAP_MESSAGE_CON;
- pdu->tid = coap_new_message_id(session);
- pdu->code = code;
-
- if (optlist) {
- coap_add_optlist_pdu(pdu, optlist);
- }
-
- coap_add_data(pdu, ptr->length, ptr->s);
- return pdu;
-}
-
-int coap_event(struct coap_context_t *ctx, coap_event_t event, struct coap_session_t *session) {
- UNUSED(session);
- if (event == COAP_EVENT_SESSION_FAILED && global_ptrs.received_error) {
- global_ptrs.received_error(receiver, ctx, -1);
- }
- return 0;
-}
-
-void no_acknowledgement(struct coap_context_t *ctx, coap_session_t *session, coap_pdu_t *sent, coap_nack_reason_t reason, const coap_tid_t id) {
- UNUSED(session);
- UNUSED(sent);
- UNUSED(reason);
- UNUSED(id);
- if (global_ptrs.received_error) {
- global_ptrs.received_error(receiver, ctx, -1);
- }
-}
-
-void response_handler(struct coap_context_t *ctx, struct coap_session_t *session, coap_pdu_t *sent, coap_pdu_t *received, const coap_tid_t id) {
- UNUSED(session);
- UNUSED(sent);
- UNUSED(id);
- coap_opt_iterator_t opt_iter;
- coap_opt_t * block_opt = coap_check_option(received, COAP_OPTION_BLOCK1, &opt_iter);
- if (block_opt) {
- printf("Block option not currently supported");
- } else {
- if (!global_ptrs.data_received) {
- return;
- }
-
- if (COAP_RESPONSE_CLASS(received->code) == 2 || received->code == COAP_RESPONSE_400) {
- if (global_ptrs.data_received) {
- CoapMessage * const msg = create_coap_message(received);
- global_ptrs.data_received(receiver, ctx, msg);
- }
- } else {
- if (global_ptrs.received_error)
- global_ptrs.received_error(receiver, ctx, received->code);
- }
- }
-}
-
-int resolve_address(const struct coap_str_const_t *server, struct sockaddr *destination) {
- struct addrinfo *result;
- struct addrinfo *iterative_obj;
- struct addrinfo hints;
- static char addrstr[256];
- int error, len = -1;
-
- memset(addrstr, 0, sizeof(addrstr));
- if (server->length)
- memcpy(addrstr, server->s, server->length);
- else
- memcpy(addrstr, "127.0.0.1", 9);
-
- memset((char *) &hints, 0, sizeof(hints));
- hints.ai_socktype = SOCK_DGRAM;
- hints.ai_family = AF_UNSPEC;
-
- error = getaddrinfo(addrstr, NULL, &hints, &result);
-
- if (error != 0) {
- perror("getaddrinfo");
-
- return error;
- }
-
- for (iterative_obj = result; iterative_obj != NULL; iterative_obj = iterative_obj->ai_next) {
- switch (iterative_obj->ai_family) {
- case AF_INET6:
- case AF_INET:
- len = iterative_obj->ai_addrlen;
- memcpy(destination, iterative_obj->ai_addr, len);
- freeaddrinfo(result);
- return len;
- default:
- ;
- }
- }
-
- freeaddrinfo(result);
- return len;
-}
-
-#ifdef __cplusplus
-}
-#endif
-
diff --git a/extensions/coap/nanofi/coap_functions.h b/extensions/coap/nanofi/coap_functions.h
deleted file mode 100644
index 8e3b2c1fa..000000000
--- a/extensions/coap/nanofi/coap_functions.h
+++ /dev/null
@@ -1,126 +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 EXTENSIONS_COAP_NANOFI_COAP_FUNCTIONS_H_
-#define EXTENSIONS_COAP_NANOFI_COAP_FUNCTIONS_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-
-typedef unsigned char method_t;
-
-#include "coap2/coap.h"
-#include "coap2/uri.h"
-#include "coap2/address.h"
-
-#ifdef WIN32
-#include <winsock2.h>
-#else
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <netdb.h>
-#endif
-
-#include "coap_message.h"
-
-
-typedef struct {
- void (*data_received)(void *receiver_context, struct coap_context_t *ctx, CoapMessage *const);
- void (*received_error)(void *receiver_context, struct coap_context_t *ctx, unsigned int code);
-} callback_pointers;
-
-/**
- * Initialize the API access. Not thread safe.
- */
-void init_coap_api(void *rcvr, callback_pointers *ptrs);
-
-/**
- * Creates a CoAP session. Provide it double pointer to the context and session to instantiate those structs.
- * @param ctx coap context
- * @param session coap session
- * @param node device node name
- * @param dst_addr destination address
- * @return 0 if success -1 otherwise.
- */
-int create_session(coap_context_t **ctx, coap_session_t **session, const char *node, const char *port, coap_address_t *dst_addr);
-
-/**
- * Creates an endpoint context
- * @param ctx pointer to a context
- * @param node device node name
- * @param port device port
- */
-int create_endpoint_context(coap_context_t **ctx, const char *node, const char *port);
-
-/**
- * Creates a request object with an already allocated context and session. The option list is sent in as an arry ptr.
- * @param ctx coap context
- * @param session coap session
- * @param optlist option list array
- * @param code coap request code
- * @param ptr ptr to the coap payload
- * @returns pointer to a newly formed PDU ( protocol data unit )
- */
-struct coap_pdu_t *create_request(struct coap_context_t *ctx, struct coap_session_t *session, coap_optlist_t **optlist, unsigned char code, coap_str_const_t *ptr);
-
-/**
- * Function can be used to receive coap events
- * @param ctx context that performed event
- * @param event coap event launched
- * @param session session that performed event.
- * @return 0 as a success ( events in this library aren't noteworthy to PDU )
- */
-int coap_event(struct coap_context_t *ctx, coap_event_t event, struct coap_session_t *session);
-
-/**
- * Function can be used when a noack is received from the library
- * @param ctx context that performed event
- * @param session session that performed event.
- * @parma sent pdu that was sent
- * @param reason reason PDU was not acked
- * @param event coap event launched
- * @param id id of ack
- */
-void no_acknowledgement(struct coap_context_t *ctx, coap_session_t *session, coap_pdu_t *sent, coap_nack_reason_t reason, const coap_tid_t id);
-
-/**
- * Responser handler is the function launched when data is received
- * @param ctx context
- * @param session coap session
- * @param sent PDU that was sent
- * @param received PDU that was received
- * @param id id of ack
- */
-void response_handler(struct coap_context_t *ctx, struct coap_session_t *session, coap_pdu_t *sent, coap_pdu_t *received, const coap_tid_t id);
-
-/**
- * Resolves the destination address of the server and places that into dst
- * @param server server to connect to
- * @param dst destination pointer
- * @return 0 if sucess -1 otherwise
- */
-int resolve_address(const struct coap_str_const_t *server, struct sockaddr *destination);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /* EXTENSIONS_COAP_NANOFI_COAP_FUNCTIONS_H_ */
-
diff --git a/extensions/coap/nanofi/coap_message.c b/extensions/coap/nanofi/coap_message.c
deleted file mode 100644
index 72c3d5c56..000000000
--- a/extensions/coap/nanofi/coap_message.c
+++ /dev/null
@@ -1,29 +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.
- */
-#include "coap_message.h"
-
-CoapMessage* create_coap_message(const coap_pdu_t * const pdu) {
- CoapMessage *message = (CoapMessage*) malloc(sizeof(CoapMessage));
- coap_get_data(pdu, &message->size_, &message->data_);
- message->code_ = pdu->code;
- return message;
-}
-
-void free_coap_message(CoapMessage *msg) {
- free(msg);
-}
diff --git a/extensions/coap/nanofi/coap_message.h b/extensions/coap/nanofi/coap_message.h
deleted file mode 100644
index 65e0b808b..000000000
--- a/extensions/coap/nanofi/coap_message.h
+++ /dev/null
@@ -1,52 +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.
- */
-#pragma once
-
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <coap2/coap.h>
-#include <stdint.h>
-#include <stdio.h>
-
-
-
-/**
- * CoAP-2 in libcoap uses uint8_t * while the first version uses a different type, so we will have to cast
- * the data. We have to keep this in mind with the API that we use.
- */
-typedef struct {
- uint32_t code_;
- size_t size_;
- uint8_t *data_;
-} CoapMessage;
-
-/**
- * Create a new CoAMessage, taking ownership of the aforementioned buffers
- */
-CoapMessage* create_coap_message(const coap_pdu_t * const pdu);
-/**
- * FRee the CoAP messages that are provided.
- */
-void free_coap_message(CoapMessage *msg);
-
-#ifdef __cplusplus
-}
-#endif
diff --git a/extensions/coap/nanofi/coap_server.c b/extensions/coap/nanofi/coap_server.c
deleted file mode 100644
index 4f43a6d2c..000000000
--- a/extensions/coap/nanofi/coap_server.c
+++ /dev/null
@@ -1,78 +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.
- */
-#include "coap_server.h"
-#include "coap_functions.h"
-/**
- * Create a new CoAPServer
- */
-CoapServerContext* create_server(const char * const server_hostname, const char * const port) {
- CoapServerContext *server = (CoapServerContext*) malloc(sizeof(CoapServerContext));
- memset(server, 0x00, sizeof(CoapServerContext));
- if (create_endpoint_context(&server->ctx, server_hostname, port)) {
- free_server(server);
- }
-
- return server;
-}
-
-CoapEndpoint* create_endpoint(CoapServerContext * const server, const char * const resource_path, uint8_t method, coap_method_handler_t handler) {
- CoapEndpoint *endpoint = (CoapEndpoint*) malloc(sizeof(CoapEndpoint));
- memset(endpoint, 0x00, sizeof(CoapEndpoint));
- endpoint->server = server;
- int8_t flags = COAP_RESOURCE_FLAGS_NOTIFY_CON;
- coap_str_const_t *path = NULL;
- if (NULL != resource_path) {
- path = coap_new_str_const((const uint8_t *) resource_path, strlen(resource_path));
- }
- endpoint->resource = coap_resource_init(path, flags);
- coap_add_attr(endpoint->resource, coap_make_str_const("title"), coap_make_str_const("\"Created CoapEndpoint\""), 0);
- if ( add_endpoint(endpoint, method, handler) ) {
- return 0x00;
- }
- coap_add_resource(server->ctx, endpoint->resource);
- if (path) {
- coap_delete_str_const(path);
- }
- return endpoint;
-}
-
-int8_t add_endpoint(CoapEndpoint * const endpoint, uint8_t method, coap_method_handler_t handler) {
- if (endpoint == NULL || handler == NULL)
- return -1;
-
- coap_register_handler(endpoint->resource, method, handler);
- return 0;
-}
-
-/**
- * FRee the CoAP messages that are provided.
- */
-void free_endpoint(CoapEndpoint * const endpoint) {
- if (endpoint) {
- free((void*) endpoint);
- }
-}
-void free_server(CoapServerContext * const server) {
- if (server) {
- if (server->ctx) {
- coap_delete_all_resources(server->ctx);
- coap_free_context(server->ctx);
- }
- free(server);
- }
-}
diff --git a/extensions/coap/nanofi/coap_server.h b/extensions/coap/nanofi/coap_server.h
deleted file mode 100644
index cd49f8b2d..000000000
--- a/extensions/coap/nanofi/coap_server.h
+++ /dev/null
@@ -1,80 +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.
- */
-#pragma once
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <coap2/coap.h>
-
-#include <stdint.h>
-#include <stdio.h>
-
-/**
- * CoAP-2 in libcoap uses uint8_t * while the first version uses a different type, so we will have to cast
- * the data. We have to keep this in mind with the API that we use.
- */
-typedef struct {
- struct coap_context_t* ctx;
- coap_address_t src_addr;
- coap_optlist_t *optlist;
-} CoapServerContext;
-
-
-typedef struct {
- CoapServerContext *server;
- coap_resource_t *resource;
-} CoapEndpoint;
-
-
-/**
- * Create a new CoAPServer using the host name provide
- * @param server_hostname hostname
- * @param port port requested
- * @param title title of base resource
- * @return CoAPServer structure.
- */
-CoapServerContext* create_server(const char *const server_hostname, const char * const port);
-
-/**
- * Creates an endpoint for the provided service context
- */
-CoapEndpoint* create_endpoint(CoapServerContext * const, const char * const resource_path, uint8_t method, coap_method_handler_t handler);
-
-/**
- * Adds an endpoint to the provided CoapEndpoint structure
- * @param endpoint endpoint we are adding the handler to
- * @method method we're adding for CoAP messages
- * @param handler handler we're using for the endpoint.
- */
-int8_t add_endpoint(CoapEndpoint * const endpoint, uint8_t method, coap_method_handler_t handler);
-
-
-/**
- * Free the CoAP messages that are provided.
- */
-void free_endpoint(CoapEndpoint * const);
-/**
- * Frees the CoAP Server context.
- */
-void free_server(CoapServerContext * const);
-
-#ifdef __cplusplus
-}
-#endif
diff --git a/extensions/coap/protocols/CoapC2Protocol.cpp b/extensions/coap/protocols/CoapC2Protocol.cpp
deleted file mode 100644
index df68004f2..000000000
--- a/extensions/coap/protocols/CoapC2Protocol.cpp
+++ /dev/null
@@ -1,299 +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.
- */
-#include "CoapC2Protocol.h"
-#include "c2/PayloadSerializer.h"
-#include "c2/PayloadParser.h"
-#include "coap_functions.h"
-#include "coap_message.h"
-#include "io/OutputStream.h"
-#include "core/Resource.h"
-#include "utils/gsl.h"
-#include "utils/span.h"
-
-namespace org::apache::nifi::minifi::coap::c2 {
-
-uint8_t CoapProtocol::REGISTRATION_MSG[8] = { 0x72, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72 };
-
-CoapProtocol::CoapProtocol(std::string_view name, const utils::Identifier &uuid)
- : RESTSender(name, uuid),
- require_registration_(false) {
-}
-
-CoapProtocol::~CoapProtocol() = default;
-
-void CoapProtocol::initialize(core::controller::ControllerServiceProvider* controller, const std::shared_ptr<Configure> &configure) {
- RESTSender::initialize(controller, configure);
- if (configure->get(minifi::Configuration::nifi_c2_coap_connector_service, controller_service_name_)) {
- auto service = controller->getControllerService(controller_service_name_);
- coap_service_ = std::static_pointer_cast<coap::controllers::CoapConnectorService>(service);
- } else {
- logger_->log_info("No CoAP connector configured, so using default service");
- coap_service_ = std::make_shared<coap::controllers::CoapConnectorService>("cs", configure);
- coap_service_->onEnable();
- }
-}
-
-minifi::c2::C2Payload CoapProtocol::consumePayload(const std::string &url, const minifi::c2::C2Payload &payload, minifi::c2::Direction direction, bool /*async*/) {
- return RESTSender::consumePayload(url, payload, direction, false);
-}
-
-int CoapProtocol::writeAcknowledgement(io::OutputStream *stream, const minifi::c2::C2Payload &payload) {
- auto ident = payload.getIdentifier();
- auto state = payload.getStatus().getState();
- stream->write(ident);
- uint8_t payloadState = 0;
- switch (state) {
- case state::UpdateState::NESTED:
- case state::UpdateState::INITIATE:
- case state::UpdateState::FULLY_APPLIED:
- case state::UpdateState::READ_COMPLETE:
- case state::UpdateState::NO_OPERATION:
- payloadState = 0;
- break;
- case state::UpdateState::NOT_APPLIED:
- case state::UpdateState::PARTIALLY_APPLIED:
- payloadState = 1;
- break;
- case state::UpdateState::READ_ERROR:
- payloadState = 2;
- break;
- case state::UpdateState::SET_ERROR:
- payloadState = 3;
- break;
- }
- stream->write(&payloadState, 1);
- return 0;
-}
-
-int CoapProtocol::writeHeartbeat(io::OutputStream *stream, const minifi::c2::C2Payload &payload) {
- bool byte;
- uint16_t size = 0;
-
- logger_->log_trace("Writing heartbeat");
-
- try {
- const auto deviceIdent = minifi::c2::PayloadParser::getInstance(payload).in("deviceInfo").getAs<std::string>("identifier");
-
- const auto agentIdent = minifi::c2::PayloadParser::getInstance(payload).in("agentInfo").getAs<std::string>("identifier");
-
- stream->write(deviceIdent, false);
-
- logger_->log_trace("Writing heartbeat with device Ident {} and agent Ident {}", deviceIdent, agentIdent);
-
- if (agentIdent.empty()) {
- return -1;
- }
- stream->write(agentIdent, false);
-
- try {
- auto flowInfoParser = minifi::c2::PayloadParser::getInstance(payload).in("flowInfo");
- auto componentParser = flowInfoParser.in("components");
- auto queueParser = flowInfoParser.in("queues");
- auto vfsParser = flowInfoParser.in("versionedFlowSnapshotURI");
- byte = true;
- stream->write(byte);
- size = componentParser.getSize();
- stream->write(size);
-
- componentParser.foreach([this, stream](const minifi::c2::C2Payload &component) {
- auto myParser = minifi::c2::PayloadParser::getInstance(component);
- bool running = false;
- stream->write(component.getLabel());
- try {
- running = myParser.getAs<bool>("running");
- }
- catch(const minifi::c2::PayloadParseException &e) {
- logger_->log_error("Could not find running in components");
- }
- stream->write(running);
- });
- size = queueParser.getSize();
- stream->write(size);
- queueParser.foreach([this, stream](const minifi::c2::C2Payload &component) {
- auto myParser = minifi::c2::PayloadParser::getInstance(component);
- stream->write(component.getLabel());
- uint64_t datasize = 0;
- uint64_t datasizemax = 0;
- uint64_t qsize = 0;
- uint64_t sizemax = 0;
- try {
- datasize = myParser.getAs<uint64_t>("dataSize");
- datasizemax = myParser.getAs<uint64_t>("dataSizeMax");
- qsize = myParser.getAs<uint64_t>("size");
- sizemax = myParser.getAs<uint64_t>("sizeMax");
- }
- catch(const minifi::c2::PayloadParseException &e) {
- logger_->log_error("Could not find queue sizes");
- }
- stream->write(datasize);
- stream->write(datasizemax);
- stream->write(qsize);
- stream->write(sizemax);
- });
-
- auto bucketId = vfsParser.getAs<std::string>("bucketId");
- auto flowId = vfsParser.getAs<std::string>("flowId");
-
- stream->write(bucketId);
- stream->write(flowId);
- } catch (const minifi::c2::PayloadParseException &pe) {
- logger_->log_error("Parser exception occurred, but is ignorable, reason {}", pe.what());
- // okay to ignore
- byte = false;
- stream->write(byte);
- }
- } catch (const minifi::c2::PayloadParseException &e) {
- logger_->log_error("Parser exception occurred, reason {}", e.what());
- return -1;
- }
- return 0;
-}
-
-minifi::c2::Operation CoapProtocol::getOperation(int type) {
- switch (type) {
- case 0:
- return minifi::c2::Operation::acknowledge;
- case 1:
- return minifi::c2::Operation::heartbeat;
- case 2:
- return minifi::c2::Operation::clear;
- case 3:
- return minifi::c2::Operation::describe;
- case 4:
- return minifi::c2::Operation::restart;
- case 5:
- return minifi::c2::Operation::start;
- case 6:
- return minifi::c2::Operation::update;
- case 7:
- return minifi::c2::Operation::stop;
- case 8:
- return minifi::c2::Operation::pause;
- case 9:
- return minifi::c2::Operation::resume;
- default:
- gsl_FailFast();
- }
-}
-
-minifi::c2::C2Payload CoapProtocol::serialize(const minifi::c2::C2Payload &payload) {
- if (nullptr == coap_service_) {
- // return an error if we don't have a coap service
- logger_->log_error("CoAP service requested without any configured hostname or port");
- return {payload.getOperation(), state::UpdateState::READ_ERROR};
- }
-
- if (require_registration_) {
- logger_->log_debug("Server requested agent registration, so attempting");
- auto response = minifi::c2::RESTSender::consumePayload(rest_uri_, payload, minifi::c2::TRANSMIT, false);
- if (response.getStatus().getState() == state::UpdateState::READ_ERROR) {
- logger_->log_trace("Could not register");
- return {payload.getOperation(), state::UpdateState::READ_COMPLETE};
- } else {
- logger_->log_trace("Registered agent.");
- }
- require_registration_ = false;
-
- return {payload.getOperation(), state::UpdateState::READ_COMPLETE};
- }
-
- uint16_t version = 0;
- uint8_t payload_type = 0;
- uint16_t size = 0;
- io::BufferStream stream;
-
- stream.write(version);
- std::string endpoint = "heartbeat";
- switch (payload.getOperation()) {
- case minifi::c2::Operation::acknowledge:
- endpoint = "acknowledge";
- payload_type = 0;
- stream.write(&payload_type, 1);
- if (writeAcknowledgement(&stream, payload) != 0) {
- return {payload.getOperation(), state::UpdateState::READ_ERROR};
- }
- break;
- case minifi::c2::Operation::heartbeat:
- payload_type = 1;
- stream.write(&payload_type, 1);
- if (writeHeartbeat(&stream, payload) != 0) {
- logger_->log_error("Could not write heartbeat");
- return {payload.getOperation(), state::UpdateState::READ_ERROR};
- }
- break;
- default:
- logger_->log_error("Could not identify operation");
- return {payload.getOperation(), state::UpdateState::READ_ERROR};
- }
-
- size_t bsize = stream.size();
-
- CoapMessage msg;
- msg.data_ = const_cast<uint8_t *>(utils::as_span<const uint8_t>(stream.getBuffer()).data());
- msg.size_ = bsize;
-
- coap::controllers::CoapResponse message = coap_service_->sendPayload(COAP_REQUEST_POST, endpoint, &msg);
- const auto message_data = message.getData();
-
- if (isRegistrationMessage(message)) {
- require_registration_ = true;
- } else if (!message_data.empty()) {
- io::BufferStream responseStream(message_data);
- responseStream.read(version);
- responseStream.read(size);
- logger_->log_trace("Received ack. version {}. number of operations {}", version, size);
- minifi::c2::C2Payload new_payload(payload.getOperation(), state::UpdateState::NESTED);
- for (int i = 0; i < size; i++) {
- uint8_t operationType;
- uint16_t argsize = 0;
- std::string operand;
- std::string id;
- REQUIRE_SIZE_IF(1, responseStream.read(operationType))
- REQUIRE_VALID(responseStream.read(id, false))
- REQUIRE_VALID(responseStream.read(operand, false))
-
- logger_->log_trace("Received op {}, with id {} and operand {}", operationType, id, operand);
- auto newOp = getOperation(operationType);
- minifi::c2::C2Payload nested_payload(newOp, state::UpdateState::READ_COMPLETE);
- nested_payload.setIdentifier(id);
- minifi::c2::C2ContentResponse new_command(newOp);
- new_command.delay = 0;
- new_command.required = true;
- new_command.ttl = -1;
- new_command.name = operand;
- new_command.ident = id;
- responseStream.read(argsize);
- for (int j = 0; j < argsize; j++) {
- std::string key;
- std::string value;
- REQUIRE_VALID(responseStream.read(key))
- REQUIRE_VALID(responseStream.read(value))
- new_command.operation_arguments[key] = value;
- }
-
- nested_payload.addContent(std::move(new_command));
- new_payload.addPayload(std::move(nested_payload));
- }
- return new_payload;
- }
-
- return {payload.getOperation(), state::UpdateState::READ_ERROR};
-}
-
-REGISTER_RESOURCE(CoapProtocol, InternalResource);
-
-} // namespace org::apache::nifi::minifi::coap::c2
diff --git a/extensions/coap/protocols/CoapC2Protocol.h b/extensions/coap/protocols/CoapC2Protocol.h
deleted file mode 100644
index 145cc6139..000000000
--- a/extensions/coap/protocols/CoapC2Protocol.h
+++ /dev/null
@@ -1,126 +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.
- */
-#pragma once
-
-#include <cstring>
-#include <algorithm>
-#include <iostream>
-#include <memory>
-#include <utility>
-#include <map>
-#include <string>
-#include <vector>
-#include <stdexcept>
-#include <mutex>
-
-#include "c2/C2Protocol.h"
-#include "io/OutputStream.h"
-#include "agent/agent_version.h"
-#include "CoapConnector.h"
-
-#include "coap2/coap.h"
-#include "coap2/uri.h"
-#include "coap2/address.h"
-#include "c2/protocols/RESTSender.h"
-
-namespace org::apache::nifi::minifi::coap::c2 {
-
-#define REQUIRE_VALID(x) \
- if (io::isError(x)) { \
- return minifi::c2::C2Payload(payload.getOperation(), state::UpdateState::READ_ERROR); \
- }
-
-#define REQUIRE_SIZE_IF(y, x) \
- if (y != x) { \
- return minifi::c2::C2Payload(payload.getOperation(), state::UpdateState::READ_ERROR); \
- }
-
-/**
- * CoAP is the Constrained Application Protocol, which defines a specialized web transfer protocol that can be
- * used on devices with constrained resources.
- */
-class CoapProtocol : public minifi::c2::RESTSender {
- public:
- explicit CoapProtocol(std::string_view name, const utils::Identifier &uuid = utils::Identifier());
-
- ~CoapProtocol() override;
-
- EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 0>{};
- EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
- EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
-
- minifi::c2::C2Payload consumePayload(const std::string &url, const minifi::c2::C2Payload &payload, minifi::c2::Direction direction, bool async) override;
-
- minifi::c2::C2Payload consumePayload(const minifi::c2::C2Payload &payload, minifi::c2::Direction /*direction*/, bool /*async*/) override {
- return serialize(payload);
- }
-
- void update(const std::shared_ptr<Configure>& /*configure*/) override {
- // no op.
- }
-
- void initialize(core::controller::ControllerServiceProvider* controller, const std::shared_ptr<Configure> &configure) override;
-
- protected:
- static bool isRegistrationMessage(controllers::CoapResponse &response) {
- const auto response_data = response.getData();
- if (LIKELY(response_data.size() != 8)) {
- return false;
- }
- return response.getCode() == COAP_RESPONSE_400 && !memcmp(response_data.data(), REGISTRATION_MSG, response_data.size());
- }
-
- /**
- * Returns the operation for the translated integer
- * @param type input type
- * @return Operation
- */
- static minifi::c2::Operation getOperation(int type);
-
- /**
- * Writes a heartbeat to the provided OutputStream ptr.
- * @param stream OutputStream
- * @param payload payload to serialize
- * @return result 0 if success failure otherwise
- */
- int writeHeartbeat(io::OutputStream *stream, const minifi::c2::C2Payload &payload);
-
- /**
- * Writes a acknowledgement to the provided OutputStream ptr.
- * @param stream OutputStream
- * @param payload payload to serialize
- * @return result 0 if success failure otherwise
- */
- static int writeAcknowledgement(io::OutputStream *stream, const minifi::c2::C2Payload &payload);
-
- minifi::c2::C2Payload serialize(const minifi::c2::C2Payload &payload);
-
- std::shared_ptr<coap::controllers::CoapConnectorService> coap_service_;
-
- std::mutex protocol_mutex_;
-
- bool require_registration_;
-
- std::string controller_service_name_;
-
- private:
- static uint8_t REGISTRATION_MSG[8];
-
- std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CoapProtocol>::getLogger();
-};
-
-} // namespace org::apache::nifi::minifi::coap::c2
diff --git a/extensions/coap/server/CoapServer.cpp b/extensions/coap/server/CoapServer.cpp
deleted file mode 100644
index 8b771900d..000000000
--- a/extensions/coap/server/CoapServer.cpp
+++ /dev/null
@@ -1,44 +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.
- */
-#include "CoapServer.h"
-#include <coap2/coap.h>
-#include <map>
-#include <functional>
-
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace coap {
-
-
-std::map<coap_resource_t*, std::function<CoapResponse(CoapQuery)>> CoapServer::functions_;
-CoapServer::~CoapServer() {
- running_ = false;
- future.get();
- if (server_) {
- free_server(server_);
- }
-}
-
-
-} /* namespace coap */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
diff --git a/extensions/coap/server/CoapServer.h b/extensions/coap/server/CoapServer.h
deleted file mode 100644
index fc3c067a2..000000000
--- a/extensions/coap/server/CoapServer.h
+++ /dev/null
@@ -1,238 +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.
- */
-
-#pragma once
-
-#include <coap2/coap.h>
-#include <functional>
-#include <thread>
-#include <future>
-#include <map>
-#include <memory>
-#include <string>
-#include <utility>
-#include <atomic>
-
-#include "core/Connectable.h"
-#include "coap_server.h"
-#include "coap_message.h"
-
-namespace org::apache::nifi::minifi::coap {
-
-enum class Method {
- Get,
- Post,
- Put,
- Delete
-};
-
-/**
- * CoapQuery is the request message sent by the client
- */
-class CoapQuery {
- public:
- CoapQuery(const std::string &query, std::unique_ptr<CoapMessage, decltype(&free_coap_message)> message)
- : query_(query),
- message_(std::move(message)) {
- }
- virtual ~CoapQuery() = default;
- CoapQuery(const CoapQuery &qry) = delete;
- CoapQuery(CoapQuery &&qry) = default;
-
- private:
- std::string query_;
- std::unique_ptr<CoapMessage, decltype(&free_coap_message)> message_;
-};
-
-/**
- * Coap Response that is generated by the CoapServer to the library
- */
-class CoapResponse {
- friend class CoapQuery;
- public:
- CoapResponse(int code, std::unique_ptr<uint8_t[]> data, size_t size)
- : code_(code),
- data_(std::move(data)),
- size_(size) {
- }
-
- int getCode() const {
- return code_;
- }
- size_t getSize() const {
- return size_;
- }
-
- uint8_t* getData() const {
- return data_.get();
- }
-
- CoapResponse(const CoapResponse &qry) = delete;
- CoapResponse(CoapResponse &&qry) = default;
- CoapResponse &operator=(CoapResponse &&qry) = default;
-
- private:
- int code_;
- std::unique_ptr<uint8_t[]> data_;
- size_t size_;
-};
-
-/**
- * Wrapper for the coap server functionality using an async callback to perform
- * custom operations. Intended to be used for testing, but may provide capabilities
- * elsewhere.
- */
-class CoapServer : public core::Connectable {
- public:
- explicit CoapServer(std::string name, const utils::Identifier &uuid)
- : core::Connectable(std::move(name), uuid),
- server_(nullptr),
- port_(0) {
- // TODO(_): this allows this class to be instantiated via the the class loader
- // need to define this capability in the future.
- }
- CoapServer(const std::string &hostname, uint16_t port)
- : core::Connectable(hostname),
- hostname_(hostname),
- server_(nullptr),
- port_(port) {
- coap_startup();
- auto port_str = std::to_string(port_);
- server_ = create_server(hostname_.c_str(), port_str.c_str());
- }
-
- virtual ~CoapServer();
-
- void start() {
- running_ = true;
-
- future = std::async(std::launch::async, [&]() -> uint64_t {
- while (running_) {
- int res = coap_run_once(server_->ctx, 100);
- if (res < 0) {
- break;
- }
- coap_check_notify(server_->ctx);
- }
- return 0;
- });
- }
-
- void add_endpoint(const std::string &path, Method method, std::function<CoapResponse(CoapQuery)> functor) {
- unsigned char mthd = COAP_REQUEST_POST;
- switch (method) {
- case Method::Get:
- mthd = COAP_REQUEST_GET;
- break;
- case Method::Post:
- mthd = COAP_REQUEST_POST;
- break;
- case Method::Put:
- mthd = COAP_REQUEST_PUT;
- break;
- case Method::Delete:
- mthd = COAP_REQUEST_DELETE;
- break;
- }
- auto current_endpoint = endpoints_.find(path);
- if (current_endpoint != endpoints_.end()) {
- ::add_endpoint(current_endpoint->second, mthd, handle_response_with_passthrough);
- } else {
- CoapEndpoint * const endpoint = create_endpoint(server_, path.c_str(), mthd, handle_response_with_passthrough);
- functions_.insert(std::make_pair(endpoint->resource, functor));
- endpoints_.insert(std::make_pair(path, endpoint));
- }
- }
-
- void add_endpoint(Method method, std::function<CoapResponse(CoapQuery)> functor) {
- unsigned char mthd = COAP_REQUEST_POST;
- switch (method) {
- case Method::Get:
- mthd = COAP_REQUEST_GET;
- break;
- case Method::Post:
- mthd = COAP_REQUEST_POST;
- break;
- case Method::Put:
- mthd = COAP_REQUEST_PUT;
- break;
- case Method::Delete:
- mthd = COAP_REQUEST_DELETE;
- break;
- }
- CoapEndpoint * const endpoint = create_endpoint(server_, NULL, mthd, handle_response_with_passthrough);
- functions_.insert(std::make_pair(endpoint->resource, functor));
- endpoints_.insert(std::make_pair("", endpoint));
- }
-
- /**
- * Determines if we are connected and operating
- */
- bool isRunning() const override {
- return running_.load();
- }
-
- /**
- * Block until work is available on any input connection, or the given duration elapses
- * @param timeoutMs timeout in milliseconds
- */
- void waitForWork(uint64_t timeoutMs);
-
- void yield() override {
- }
-
- /**
- * Determines if work is available by this connectable
- * @return boolean if work is available.
- */
- bool isWorkAvailable() override {
- return true;
- }
-
- protected:
- static void handle_response_with_passthrough(coap_context_t* /*ctx*/,
- struct coap_resource_t *resource,
- coap_session_t *session,
- coap_pdu_t *request,
- coap_binary_t* /*token*/,
- coap_string_t* /*query*/,
- coap_pdu_t *response) {
- auto fx = functions_.find(resource);
- if (fx != functions_.end()) {
- auto message = create_coap_message(request);
- CoapQuery qry("", std::unique_ptr<CoapMessage, decltype(&free_coap_message)>(message, free_coap_message));
- // call the UDF
- auto udfResponse = fx->second(std::move(qry));
- response = coap_pdu_init(COAP_MESSAGE_CON, COAP_RESPONSE_CODE(udfResponse.getCode()), coap_new_message_id(session), udfResponse.getSize() + 1);
- coap_add_data(response, udfResponse.getSize(), udfResponse.getData());
- if (coap_send(session, response) == COAP_INVALID_TID) {
- printf("error while returning response");
- }
- }
- }
-
- std::future<uint64_t> future;
- std::atomic<bool> running_;
- std::string hostname_;
- CoapServerContext *server_;
- static std::map<coap_resource_t*, std::function<CoapResponse(CoapQuery)>> functions_;
- std::map<std::string, CoapEndpoint*> endpoints_;
- uint16_t port_;
-};
-
-} // namespace org::apache::nifi::minifi::coap
diff --git a/extensions/coap/tests/CMakeLists.txt b/extensions/coap/tests/CMakeLists.txt
deleted file mode 100644
index 54b143e38..000000000
--- a/extensions/coap/tests/CMakeLists.txt
+++ /dev/null
@@ -1,53 +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.
-#
-
-file(GLOB COAP_INTEGRATION_TESTS "*.cpp")
-
-SET(CURL_INT_TEST_COUNT 0)
-
-FOREACH(testfile ${COAP_INTEGRATION_TESTS})
- get_filename_component(testfilename "${testfile}" NAME_WE)
- add_minifi_executable("${testfilename}" "${testfile}")
- target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/standard-processors")
- target_include_directories(${testfilename} BEFORE PRIVATE "../")
- target_include_directories(${testfilename} BEFORE PRIVATE "../server")
- target_include_directories(${testfilename} BEFORE PRIVATE "../nanofi")
- target_include_directories(${testfilename} BEFORE PRIVATE "../protocols")
- target_include_directories(${testfilename} BEFORE PRIVATE "../controllerservice")
- target_include_directories(${testfilename} BEFORE PRIVATE "../../http-curl/")
- target_include_directories(${testfilename} BEFORE PRIVATE "../../http-curl/client/")
- target_include_directories(${testfilename} BEFORE PRIVATE "../../http-curl/processors/")
- target_include_directories(${testfilename} BEFORE PRIVATE "../../http-curl/protocols/")
- target_include_directories(${testfilename} BEFORE PRIVATE "../../http-curl/sitetosite/")
- target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/civetweb/")
- target_include_directories(${testfilename} BEFORE PRIVATE ./include)
- target_compile_definitions(${testfilename} PRIVATE TEST_RESOURCES="${TEST_RESOURCES}")
- createIntegrationTests("${testfilename}")
- target_link_libraries(${testfilename} minifi-coap)
- target_link_libraries(${testfilename} minifi-civet-extensions)
- target_link_libraries(${testfilename} minifi-standard-processors)
- target_link_libraries(${testfilename} Catch2WithMain)
- add_test(NAME "${testfilename}" COMMAND "${testfilename}")
-
- MATH(EXPR CURL_INT_TEST_COUNT "${CURL_INT_TEST_COUNT}+1")
-ENDFOREACH()
-
-message("-- Finished building ${CURL_INT_TEST_COUNT} CoAP integration test file(s)...")
-
-
diff --git a/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp b/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
deleted file mode 100644
index e0febdcbd..000000000
--- a/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
+++ /dev/null
@@ -1,180 +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.
- */
-#include <sys/stat.h>
-#include <utility>
-#include <chrono>
-#include <fstream>
-#include <memory>
-#include <string>
-#include <thread>
-#include <type_traits>
-#include <vector>
-#include <iostream>
-#include <sstream>
-#include "http/BaseHTTPClient.h"
-#include "processors/InvokeHTTP.h"
-#include "unit/TestBase.h"
-#include "utils/StringUtils.h"
-#include "core/Core.h"
-#include "core/logging/Logger.h"
-#include "core/ProcessGroup.h"
-#include "FlowController.h"
-#include "properties/Configure.h"
-#include "unit/ProvenanceTestHelper.h"
-#include "CivetServer.h"
-#include "RemoteProcessorGroupPort.h"
-#include "core/ConfigurableComponent.h"
-#include "controllers/SSLContextService.h"
-#include "c2/C2Agent.h"
-#include "protocols/RESTReceiver.h"
-#include "CoapIntegrationBase.h"
-#include "processors/LogAttribute.h"
-#include "CoapC2Protocol.h"
-#include "CoapServer.h"
-#include "concurrentqueue.h"
-#include "unit/TestUtils.h"
-#include "unit/Catch.h"
-
-class VerifyCoAPServer : public CoapIntegrationBase {
- public:
- explicit VerifyCoAPServer(bool isSecure)
- : isSecure(isSecure) {
- dir = testController.createTempDirectory();
- }
-
- void testSetup() override {
- LogTestController::getInstance().setDebug<minifi::http::HTTPClient>();
- LogTestController::getInstance().setOff<minifi::processors::InvokeHTTP>();
- LogTestController::getInstance().setDebug<minifi::c2::RESTReceiver>();
- LogTestController::getInstance().setDebug<minifi::c2::C2Agent>();
- LogTestController::getInstance().setTrace<minifi::coap::c2::CoapProtocol>();
- LogTestController::getInstance().setOff<minifi::processors::LogAttribute>();
- LogTestController::getInstance().setOff<minifi::core::ProcessSession>();
- std::fstream file;
- ss << dir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- }
-
- void cleanup() override {
- unlink(ss.str().c_str());
- CoapIntegrationBase::cleanup();
- }
-
- void runAssertions() override {
- using org::apache::nifi::minifi::test::utils::verifyLogLinePresenceInPollTime;
- REQUIRE(verifyLogLinePresenceInPollTime(std::chrono::seconds(3),
- "Received ack. version 3. number of operations 1",
- "Received ack. version 3. number of operations 0",
- "Received read error event from protocol",
- "Received op 1, with id id and operand operand"));
- }
-
- void queryRootProcessGroup(std::shared_ptr<core::ProcessGroup> pg) override {
- auto proc = pg->findProcessorByName("invoke");
- REQUIRE(proc != nullptr);
-
- const auto* const inv = dynamic_cast<minifi::processors::InvokeHTTP*>(proc);
-
- REQUIRE(inv != nullptr);
- std::string url;
- inv->getProperty(minifi::processors::InvokeHTTP::URL, url);
-
- std::string port;
- std::string scheme;
- std::string path;
-
- minifi::utils::parse_http_components(url, port, scheme, path);
- uint16_t newport = std::stoi(port) + 2;
- auto new_port_str = std::to_string(newport);
-
- server = std::make_unique<minifi::coap::CoapServer>("127.0.0.1", newport);
-
-
- server->add_endpoint(minifi::coap::Method::Post, [](minifi::coap::CoapQuery)->minifi::coap::CoapResponse {
- minifi::coap::CoapResponse response(205, nullptr, 0);
- return response;
- });
-
- {
- // valid response version 3, 0 ops
- auto data = std::unique_ptr<uint8_t[]>(new uint8_t[5] { 0x00, 0x03, 0x00, 0x01, 0x00 }); // NOLINT(cppcoreguidelines-avoid-c-arrays)
- minifi::coap::CoapResponse response(205, std::move(data), 5);
- responses.enqueue(std::move(response));
- }
-
- {
- // valid response
- auto data = std::unique_ptr<uint8_t[]>(new uint8_t[5] { 0x00, 0x03, 0x00, 0x00, 0x00 }); // NOLINT(cppcoreguidelines-avoid-c-arrays)
- minifi::coap::CoapResponse response(205, std::move(data), 5);
- responses.enqueue(std::move(response));
- }
-
- {
- // should result in valid operation
- minifi::io::BufferStream stream;
- uint16_t version = 0;
- uint16_t size = 1;
- uint8_t operation = 1;
- stream.write(version);
- stream.write(size);
- stream.write(&operation, 1);
- stream.write("id");
- stream.write("operand");
-
- auto data = std::make_unique<uint8_t[]>(stream.size()); // NOLINT(cppcoreguidelines-avoid-c-arrays)
- memcpy(data.get(), stream.getBuffer().data(), stream.getBuffer().size());
- minifi::coap::CoapResponse response(205, std::move(data), stream.size());
- responses.enqueue(std::move(response));
- }
-
- server->add_endpoint("heartbeat", minifi::coap::Method::Post, [&](minifi::coap::CoapQuery)-> minifi::coap::CoapResponse {
- if (responses.size_approx() > 0) {
- minifi::coap::CoapResponse resp(500, nullptr, 0);;
- responses.try_dequeue(resp);
- return resp;
- }
- minifi::coap::CoapResponse response(500, nullptr, 0);
- return response;
- });
- server->start();
- configuration->set(minifi::Configuration::nifi_c2_enable, "true");
- configuration->set(minifi::Configuration::nifi_c2_agent_class, "test");
- configuration->set(minifi::Configuration::nifi_c2_root_classes, "DeviceInfoNode,AgentInformation,FlowInformation,RepositoryMetrics");
- configuration->set(minifi::Configuration::nifi_c2_agent_protocol_class, "CoapProtocol");
- configuration->set(minifi::Configuration::nifi_c2_agent_coap_host, "127.0.0.1");
- configuration->set(minifi::Configuration::nifi_c2_agent_coap_port, new_port_str);
- configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_period, "10");
- }
-
- protected:
- moodycamel::ConcurrentQueue<minifi::coap::CoapResponse> responses;
- std::unique_ptr<minifi::coap::CoapServer> server;
- bool isSecure;
- std::string dir;
- std::stringstream ss;
- TestController testController;
-};
-
-TEST_CASE("CoapC2VerifyHeartbeat", "[c2test]") {
- VerifyCoAPServer harness(false);
- harness.setKeyDir(TEST_RESOURCES);
- const auto test_file_path = std::filesystem::path(TEST_RESOURCES) / "CoapC2VerifyServe.yml";
- harness.run(test_file_path);
-}
diff --git a/extensions/coap/tests/CoapIntegrationBase.h b/extensions/coap/tests/CoapIntegrationBase.h
deleted file mode 100644
index 6974143ca..000000000
--- a/extensions/coap/tests/CoapIntegrationBase.h
+++ /dev/null
@@ -1,123 +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.
- */
-#pragma once
-
-#include <memory>
-#include <optional>
-#include <string>
-#include <utility>
-#include <vector>
-
-#include "integration/TestServer.h"
-#include "CivetServer.h"
-#include "integration/IntegrationBase.h"
-#include "core/yaml/YamlConfiguration.h"
-
-using namespace std::literals::chrono_literals;
-
-int log_message(const struct mg_connection* /*conn*/, const char *message) {
- puts(message);
- return 1;
-}
-
-int ssl_enable(void *, void *) {
- return 0;
-}
-
-class CoapIntegrationBase : public minifi::test::IntegrationBase {
- public:
- explicit CoapIntegrationBase(std::chrono::seconds waitTime = 5s)
- : IntegrationBase(waitTime),
- server(nullptr) {
- }
-
- void setUrl(std::string url, CivetHandler *handler);
-
- void shutdownBeforeFlowController() override {
- server.reset();
- }
-
- void run(const std::optional<std::filesystem::path>& test_file_location = {}, const std::optional<std::filesystem::path>& = {}) override {
- testSetup();
-
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>();
- std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
-
- if (test_file_location) {
- configuration->set(minifi::Configure::nifi_flow_configuration_file, test_file_location->string());
- }
- configuration->set(minifi::Configure::nifi_c2_agent_heartbeat_period, "200");
-
- std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
- content_repo->initialize(configuration);
-
- auto configuration_context = core::ConfigurationContext{
- .flow_file_repo = test_repo,
- .content_repo = content_repo,
- .configuration = configuration,
- .path = test_file_location,
- .filesystem = std::make_shared<utils::file::FileSystem>(),
- .sensitive_properties_encryptor = utils::crypto::EncryptionProvider{utils::crypto::XSalsa20Cipher{utils::crypto::XSalsa20Cipher::generateKey()}}
- };
- auto yaml_ptr = std::make_shared<core::YamlConfiguration>(configuration_context);
- core::YamlConfiguration yaml_config(configuration_context);
-
- std::shared_ptr<core::ProcessGroup> pg{ yaml_config.getRoot() };
-
- queryRootProcessGroup(pg);
-
- std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repo_metric_sources{test_repo, test_flow_repo, content_repo};
- auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, repo_metric_sources, yaml_ptr);
- auto controller = std::make_unique<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, std::move(metrics_publisher_store));
-
- controller->load();
- controller->start();
-
- runAssertions();
-
- shutdownBeforeFlowController();
- controller->waitUnload(wait_time_);
-
- cleanup();
- }
-
- protected:
- std::unique_ptr<TestServer> server;
-};
-
-void CoapIntegrationBase::setUrl(std::string url, CivetHandler *handler) {
- std::string path;
- minifi::utils::parse_http_components(url, port, scheme, path);
- CivetCallbacks callback{};
- if (url.find("localhost") != std::string::npos) {
- if (server != nullptr) {
- server->addHandler(path, handler);
- return;
- }
- if (scheme == "https" && !key_dir.empty()) {
- std::string cert;
- cert = key_dir.string() + "nifi-cert.pem";
- callback.init_ssl = ssl_enable;
- port += "s";
- callback.log_message = log_message;
- server = std::make_unique<TestServer>(port, path, handler, &callback, cert, cert);
- } else {
- server = std::make_unique<TestServer>(port, path, handler);
- }
- }
-}
diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
index 6c721dfc4..bba8bbe21 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#define EXTENSION_LIST "*minifi-*,!*coap*,!*splunk*,!*elastic*,!*grafana-loki*" // NOLINT(cppcoreguidelines-macro-usage)
+#define EXTENSION_LIST "*minifi-*,!*splunk*,!*elastic*,!*grafana-loki*" // NOLINT(cppcoreguidelines-macro-usage)
#include <cstdio>
#include <utility>
diff --git a/libminifi/include/c2/C2Agent.h b/libminifi/include/c2/C2Agent.h
index 4c6254967..63829fd2c 100644
--- a/libminifi/include/c2/C2Agent.h
+++ b/libminifi/include/c2/C2Agent.h
@@ -64,10 +64,6 @@ class C2Agent : public state::UpdateController {
std::shared_ptr<utils::file::FileSystem> filesystem,
std::function<void()> request_restart);
- ~C2Agent() noexcept override {
- delete protocol_.load();
- }
-
void initialize(core::controller::ControllerServiceProvider *controller, state::Pausable *pause_handler, state::StateMonitor* update_sink);
void start() override;
void stop() override;
@@ -222,7 +218,7 @@ class C2Agent : public state::UpdateController {
std::vector<std::unique_ptr<C2Trigger>> triggers_;
- std::atomic<C2Protocol*> protocol_{};
+ std::unique_ptr<C2Protocol> protocol_;
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<C2Agent>::getLogger();
diff --git a/libminifi/include/properties/Configuration.h b/libminifi/include/properties/Configuration.h
index 79ef4c53c..d199d75c3 100644
--- a/libminifi/include/properties/Configuration.h
+++ b/libminifi/include/properties/Configuration.h
@@ -108,13 +108,9 @@ class Configuration : public Properties {
static constexpr const char *nifi_c2_flow_url = "nifi.c2.flow.url";
static constexpr const char *nifi_c2_flow_base_url = "nifi.c2.flow.base.url";
static constexpr const char *nifi_c2_full_heartbeat = "nifi.c2.full.heartbeat";
- static constexpr const char *nifi_c2_coap_connector_service = "nifi.c2.coap.connector.service";
static constexpr const char *nifi_c2_agent_heartbeat_period = "nifi.c2.agent.heartbeat.period";
static constexpr const char *nifi_c2_agent_class = "nifi.c2.agent.class";
static constexpr const char *nifi_c2_agent_heartbeat_reporter_classes = "nifi.c2.agent.heartbeat.reporter.classes";
- static constexpr const char *nifi_c2_agent_coap_host = "nifi.c2.agent.coap.host";
- static constexpr const char *nifi_c2_agent_coap_port = "nifi.c2.agent.coap.port";
- static constexpr const char *nifi_c2_agent_protocol_class = "nifi.c2.agent.protocol.class";
static constexpr const char *nifi_c2_agent_identifier = "nifi.c2.agent.identifier";
static constexpr const char *nifi_c2_agent_identifier_fallback = "nifi.c2.agent.identifier.fallback";
static constexpr const char *nifi_c2_agent_trigger_classes = "nifi.c2.agent.trigger.classes";
diff --git a/libminifi/src/Configuration.cpp b/libminifi/src/Configuration.cpp
index 3cf4e06c2..ede730bb9 100644
--- a/libminifi/src/Configuration.cpp
+++ b/libminifi/src/Configuration.cpp
@@ -78,13 +78,9 @@ const std::unordered_map<std::string_view, gsl::not_null<const core::PropertyVal
{Configuration::nifi_c2_flow_url, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_flow_base_url, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_full_heartbeat, gsl::make_not_null(&core::StandardPropertyTypes::BOOLEAN_TYPE)},
- {Configuration::nifi_c2_coap_connector_service, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_agent_heartbeat_period, gsl::make_not_null(&core::StandardPropertyTypes::UNSIGNED_INT_TYPE)},
{Configuration::nifi_c2_agent_heartbeat_reporter_classes, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_agent_class, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
- {Configuration::nifi_c2_agent_coap_host, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
- {Configuration::nifi_c2_agent_coap_port, gsl::make_not_null(&core::StandardPropertyTypes::PORT_TYPE)},
- {Configuration::nifi_c2_agent_protocol_class, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_agent_identifier, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_agent_identifier_fallback, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
{Configuration::nifi_c2_agent_trigger_classes, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)},
diff --git a/libminifi/src/c2/C2Agent.cpp b/libminifi/src/c2/C2Agent.cpp
index 318f3e95e..5f1097fb4 100644
--- a/libminifi/src/c2/C2Agent.cpp
+++ b/libminifi/src/c2/C2Agent.cpp
@@ -45,6 +45,7 @@
#include "io/StreamPipe.h"
#include "utils/Id.h"
#include "c2/C2Utils.h"
+#include "c2/protocols/RESTSender.h"
using namespace std::literals::chrono_literals;
@@ -144,37 +145,14 @@ void C2Agent::checkTriggers() {
}
}
void C2Agent::configure(const std::shared_ptr<Configure> &configure, bool reconfigure) {
- std::string clazz;
- std::string heartbeat_period;
- std::string device;
-
if (!reconfigure) {
- if (!configure->get(Configuration::nifi_c2_agent_protocol_class, "c2.agent.protocol.class", clazz)) {
- clazz = "RESTSender";
- }
- logger_->log_info("Class is {}", clazz);
-
- auto protocol = core::ClassLoader::getDefaultClassLoader().instantiateRaw(clazz, clazz);
- if (protocol == nullptr) {
- logger_->log_warn("Class {} not found", clazz);
- protocol = core::ClassLoader::getDefaultClassLoader().instantiateRaw("RESTSender", "RESTSender");
- if (!protocol) {
- constexpr const char* errmsg = "Attempted to load RESTSender. To enable C2, please specify an active protocol for this agent.";
- logger_->log_error("{}", errmsg);
- throw Exception{ GENERAL_EXCEPTION, errmsg };
- }
-
- logger_->log_info("Class is RESTSender");
- }
-
- // Since !reconfigure, the call comes from the ctor and protocol_ is null, therefore no delete is necessary
- protocol_.exchange(dynamic_cast<C2Protocol *>(protocol));
-
- protocol_.load()->initialize(controller_, configuration_);
+ protocol_ = std::make_unique<RESTSender>("RESTSender");
+ protocol_->initialize(controller_, configuration_);
} else {
- protocol_.load()->update(configure);
+ protocol_->update(configure);
}
+ std::string heartbeat_period;
if (configure->get(Configuration::nifi_c2_agent_heartbeat_period, "c2.agent.heartbeat.period", heartbeat_period)) {
try {
if (auto heartbeat_period_ms = utils::timeutils::StringToDuration<std::chrono::milliseconds>(heartbeat_period)) {
@@ -245,7 +223,7 @@ void C2Agent::performHeartBeat() {
serializeMetrics(child_metric_payload, metric.name, metric.serialized_nodes, metric.is_array);
payload.addPayload(std::move(child_metric_payload));
}
- C2Payload response = protocol_.load()->consumePayload(payload);
+ C2Payload response = protocol_->consumePayload(payload);
enqueue_c2_server_response(std::move(response));
@@ -350,7 +328,7 @@ void C2Agent::handle_c2_server_response(const C2ContentResponse &resp) {
case Operation::restart: {
update_sink_->stop();
C2Payload response(Operation::acknowledge, resp.ident, true);
- protocol_.load()->consumePayload(response);
+ protocol_->consumePayload(response);
restart_needed_ = true;
}
break;
@@ -713,7 +691,7 @@ void C2Agent::handle_transfer(const C2ContentResponse &resp) {
std::map<std::string, std::unique_ptr<io::InputStream>> files = update_sink_->getDebugInfo();
auto bundle = bundleDebugInfo(files);
- C2Payload &&response = protocol_.load()->consumePayload(url.value(), bundle, TRANSMIT, false);
+ C2Payload &&response = protocol_->consumePayload(url.value(), bundle, TRANSMIT, false);
if (response.getStatus().getState() == state::UpdateState::READ_ERROR) {
throw C2DebugBundleError("Error while uploading");
}
@@ -724,7 +702,7 @@ void C2Agent::handle_transfer(const C2ContentResponse &resp) {
utils::TaskRescheduleInfo C2Agent::produce() {
// place priority on messages to send to the c2 server
- if (protocol_.load() != nullptr) {
+ if (protocol_ != nullptr) {
std::vector<C2Payload> payload_batch;
payload_batch.reserve(max_c2_responses);
auto getRequestPayload = [&payload_batch] (C2Payload&& payload) { payload_batch.emplace_back(std::move(payload)); };
@@ -738,7 +716,7 @@ utils::TaskRescheduleInfo C2Agent::produce() {
std::make_move_iterator(payload_batch.end()),
[&] (C2Payload&& payload) {
try {
- C2Payload response = protocol_.load()->consumePayload(payload);
+ C2Payload response = protocol_->consumePayload(payload);
enqueue_c2_server_response(std::move(response));
}
catch(const std::exception &e) {
@@ -824,14 +802,14 @@ std::optional<std::string> C2Agent::resolveUrl(const std::string& url) const {
}
std::optional<std::string> C2Agent::fetchFlow(const std::string& uri) const {
- if (!utils::string::startsWith(uri, "http") || protocol_.load() == nullptr) {
+ if (!utils::string::startsWith(uri, "http") || protocol_ == nullptr) {
// try to open the file
auto content = filesystem_->read(uri);
if (content) {
return content;
}
}
- if (protocol_.load() == nullptr) {
+ if (protocol_ == nullptr) {
logger_->log_error("Couldn't open '{}' as file and we have no protocol to request the file from", uri);
return {};
}
@@ -841,7 +819,7 @@ std::optional<std::string> C2Agent::fetchFlow(const std::string& uri) const {
return std::nullopt;
}
- C2Payload response = protocol_.load()->fetch(resolved_url.value(), update_sink_->getSupportedConfigurationFormats());
+ C2Payload response = protocol_->fetch(resolved_url.value(), update_sink_->getSupportedConfigurationFormats());
return response.getRawDataAsString();
}
@@ -993,7 +971,7 @@ void C2Agent::handleAssetUpdate(const C2ContentResponse& resp) {
return;
}
- C2Payload file_response = protocol_.load()->fetch(url);
+ C2Payload file_response = protocol_->fetch(url);
if (file_response.getStatus().getState() != state::UpdateState::READ_COMPLETE) {
send_error("Failed to fetch asset from '" + url + "'");
diff --git a/libminifi/test/integration/C2NullConfiguration.cpp b/libminifi/test/integration/C2NullConfiguration.cpp
index ad36cc807..5b1ee626d 100644
--- a/libminifi/test/integration/C2NullConfiguration.cpp
+++ b/libminifi/test/integration/C2NullConfiguration.cpp
@@ -54,9 +54,7 @@ class VerifyC2Server : public HTTPIntegrationBase {
}
void runAssertions() override {
- REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::milliseconds(wait_time_),
- "C2Agent] [error] Could not instantiate null",
- "Class is RESTSender"));
+ REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(std::chrono::milliseconds(wait_time_), "C2Agent] [error] Could not instantiate null"));
}
void queryRootProcessGroup(std::shared_ptr<core::ProcessGroup> pg) override {
@@ -75,7 +73,6 @@ class VerifyC2Server : public HTTPIntegrationBase {
minifi::utils::parse_http_components(url, port, scheme, path);
configuration->set(Configuration::nifi_c2_enable, "true");
configuration->set(Configuration::nifi_c2_agent_class, "test");
- configuration->set(Configuration::nifi_c2_agent_protocol_class, "RESTSender");
configuration->set(Configuration::nifi_c2_rest_url, "");
configuration->set(Configuration::nifi_c2_rest_url_ack, "");
configuration->set(Configuration::nifi_c2_agent_heartbeat_reporter_classes, "null");
diff --git a/libminifi/test/integration/C2PropertiesUpdateTests.cpp b/libminifi/test/integration/C2PropertiesUpdateTests.cpp
index 74326571b..7669f250d 100644
--- a/libminifi/test/integration/C2PropertiesUpdateTests.cpp
+++ b/libminifi/test/integration/C2PropertiesUpdateTests.cpp
@@ -109,7 +109,6 @@ class VerifyPropertyUpdate : public HTTPIntegrationBase {
static const std::string properties_file =
"nifi.property.one=tree\n"
- "nifi.c2.agent.protocol.class=RESTSender\n"
"nifi.c2.enable=true\n"
"nifi.c2.agent.class=test\n"
"nifi.c2.agent.heartbeat.period=500\n";
diff --git a/libminifi/test/integration/C2RequestClassTest.cpp b/libminifi/test/integration/C2RequestClassTest.cpp
index b00b0a815..ac55d7764 100644
--- a/libminifi/test/integration/C2RequestClassTest.cpp
+++ b/libminifi/test/integration/C2RequestClassTest.cpp
@@ -77,7 +77,6 @@ class VerifyC2ClassRequest : public VerifyC2Base {
explicit VerifyC2ClassRequest(std::function<bool()> verify) : verify_(std::move(verify)) {}
void configureC2() override {
- configuration->set(minifi::Configuration::nifi_c2_agent_protocol_class, "RESTSender");
configuration->set(minifi::Configuration::nifi_c2_enable, "true");
configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_period, "100");
configuration->set(minifi::Configuration::nifi_c2_root_classes, "DeviceInfoNode,AgentInformation,FlowInformation");
diff --git a/libminifi/test/integration/C2UpdateAssetTest.cpp b/libminifi/test/integration/C2UpdateAssetTest.cpp
index 9b279d0b5..71cd0b737 100644
--- a/libminifi/test/integration/C2UpdateAssetTest.cpp
+++ b/libminifi/test/integration/C2UpdateAssetTest.cpp
@@ -74,7 +74,6 @@ class C2HeartbeatHandler : public HeartbeatHandler {
class VerifyC2AssetUpdate : public VerifyC2Base {
public:
void configureC2() override {
- configuration->set("nifi.c2.agent.protocol.class", "RESTSender");
configuration->set("nifi.c2.enable", "true");
configuration->set("nifi.c2.agent.heartbeat.period", "100");
}
diff --git a/libminifi/test/integration/C2VerifyServeResults.cpp b/libminifi/test/integration/C2VerifyServeResults.cpp
index a4c51847a..7da378412 100644
--- a/libminifi/test/integration/C2VerifyServeResults.cpp
+++ b/libminifi/test/integration/C2VerifyServeResults.cpp
@@ -75,7 +75,6 @@ class VerifyC2Server : public HTTPIntegrationBase {
configuration->set(minifi::Configuration::nifi_c2_enable, "true");
configuration->set(minifi::Configuration::nifi_c2_agent_class, "test");
configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_reporter_classes, "RESTReceiver");
- configuration->set(minifi::Configuration::nifi_c2_agent_protocol_class, "RESTSender");
configuration->set(minifi::Configuration::nifi_c2_rest_listener_port, port);
configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_period, "10");
}
diff --git a/libminifi/test/libtest/integration/HTTPIntegrationBase.h b/libminifi/test/libtest/integration/HTTPIntegrationBase.h
index 43e71c573..42a534485 100644
--- a/libminifi/test/libtest/integration/HTTPIntegrationBase.h
+++ b/libminifi/test/libtest/integration/HTTPIntegrationBase.h
@@ -72,7 +72,6 @@ class VerifyC2Base : public HTTPIntegrationBase {
}
void configureC2() override {
- configuration->set(minifi::Configuration::nifi_c2_agent_protocol_class, "RESTSender");
configuration->set(minifi::Configuration::nifi_c2_enable, "true");
configuration->set(minifi::Configuration::nifi_c2_agent_class, "test");
configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_period, "1000");
@@ -124,7 +123,6 @@ class VerifyC2Update : public HTTPIntegrationBase {
}
void configureC2() override {
- configuration->set(minifi::Configuration::nifi_c2_agent_protocol_class, "RESTSender");
configuration->set(minifi::Configuration::nifi_c2_enable, "true");
configuration->set(minifi::Configuration::nifi_c2_agent_class, "test");
configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_period, "1000");
@@ -152,7 +150,6 @@ class VerifyFlowFetched : public HTTPIntegrationBase {
}
void configureC2() override {
- configuration->set(minifi::Configuration::nifi_c2_agent_protocol_class, "RESTSender");
configuration->set(minifi::Configuration::nifi_c2_enable, "true");
configuration->set(minifi::Configuration::nifi_c2_agent_class, "test");
configuration->set(minifi::Configuration::nifi_c2_agent_heartbeat_period, "1000");
diff --git a/libminifi/test/resources/encrypted.minifi.properties b/libminifi/test/resources/encrypted.minifi.properties
index 3a9b84452..77e952170 100644
--- a/libminifi/test/resources/encrypted.minifi.properties
+++ b/libminifi/test/resources/encrypted.minifi.properties
@@ -52,13 +52,6 @@ nifi.rest.api.password.protected=xsalsa20poly1305
## Enabling C2 Uncomment each of the following options
## define those with missing options
nifi.c2.enable=true
-## define protocol parameters
-## The default is RESTSender.
-## Alternatively, you may use CoapProtocol if that extension is built.
-nifi.c2.agent.protocol.class=RESTSender
-#nifi.c2.agent.protocol.class=CoapProtocol
-#nifi.c2.agent.coap.host=
-#nifi.c2.agent.coap.port=
## base URL of the c2 server,
## very likely the same base url of rest urls
nifi.c2.flow.base.url=http://localhost:10080/c2-server/api
diff --git a/msi/WixWin.wsi.in b/msi/WixWin.wsi.in
index 1b38e9270..09920256a 100644
--- a/msi/WixWin.wsi.in
+++ b/msi/WixWin.wsi.in
@@ -250,21 +250,6 @@ ${WIX_EXTRA_FEATURES}
<Condition Action="disable"><![CDATA[ENABLEC2<>1]]></Condition>
<Condition Action="enable">ENABLEC2="1"</Condition>
</Control>
- <!--
- comment when available
- <Control Id="AgentProtocolLabel" Type="Text" X="20" Y="240" Width="60" Height="20" NoPrefix="yes" Text="Agent Protocol" />
- <Control Id="AgentProtocolComboBreaker" Type="ComboBox" X="20" Y="260" Width="300" Height="17" Property="AGENT_PROTOCOL" >
-
-
-
- <ComboBox Property="AGENT_PROTOCOL">
- <ListItem Value="RESTSender" />
- <ListItem Value="COAPProtocol" />
- </ComboBox>
- <Condition Action="disable"><![CDATA[ENABLEC2<>1]]></Condition>
- <Condition Action="enable">ENABLEC2="1"</Condition>
- </Control>
--->
</Dialog>
<UIRef Id="WixUI_HK" />
@@ -273,7 +258,6 @@ ${WIX_EXTRA_FEATURES}
<Property Id="AGENT_CLASS" Value="Your Agent Class" />
<Property Id="AGENT_IDENTIFIER" />
- <Property Id="AGENT_PROTOCOL" Value="RESTSender" />
<Property Id="AGENT_HEARTBEAT" Value="30 sec" />
<Property Id="SERVER_HEARTBEAT" Value="http://localhost:8181/heartbeat" />
<Property Id="SERVER_ACK" Value="http://localhost:8181/acknowledge" />
@@ -322,7 +306,6 @@ ${WIX_EXTRA_FEATURES}
<IniFile Id="ConfigFileA" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.agent.class" Value="[AGENT_CLASS]" />
<IniFile Id="ConfigFileI" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.agent.identifier" Value="[AGENT_IDENTIFIER]" />
<IniFile Id="ConfigFileE" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.enable" Value="true" />
- <IniFile Id="ConfigFileP" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.agent.protocol.class" Value="[AGENT_PROTOCOL]" />
<IniFile Id="ConfigFileT" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.agent.heartbeat.period" Value="[AGENT_HEARTBEAT]" />
<IniFile Id="ConfigFileH" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.rest.url" Value="[SERVER_HEARTBEAT]" />
<IniFile Id="ConfigFileAck" Action="addLine" Name="minifi.properties" Directory="CONFIGDIR" Section="c2props" Key="nifi.c2.rest.url.ack" Value="[SERVER_ACK]" />
diff --git a/thirdparty/libcoap/libcoap-windows-cmake.patch b/thirdparty/libcoap/libcoap-windows-cmake.patch
deleted file mode 100644
index 9150cd0ee..000000000
--- a/thirdparty/libcoap/libcoap-windows-cmake.patch
+++ /dev/null
@@ -1,93 +0,0 @@
-diff -rupN orig/CMakeLists.txt patched/CMakeLists.txt
---- orig/CMakeLists.txt 1970-01-01 01:00:00.000000000 +0100
-+++ patched/CMakeLists.txt 2019-11-21 17:42:30.000000000 +0100
-@@ -0,0 +1,77 @@
-+cmake_minimum_required(VERSION 3.10)
-+
-+project(libcoap)
-+
-+set(SOURCES src/address.c
-+ src/async.c
-+ src/block.c
-+ src/coap_debug.c
-+ src/coap_event.c
-+ src/coap_hashkey.c
-+ src/coap_gnutls.c
-+ src/coap_io.c
-+ src/coap_notls.c
-+ src/coap_openssl.c
-+ src/coap_session.c
-+ src/coap_time.c
-+ src/coap_tinydtls.c
-+ src/encode.c
-+ src/mem.c
-+ src/net.c
-+ src/option.c
-+ src/pdu.c
-+ src/resource.c
-+ src/str.c
-+ src/subscribe.c
-+ src/uri.c)
-+if(WIN32)
-+ configure_file(include/coap2/coap.h.windows include/coap2/coap.h COPYONLY)
-+ configure_file(coap_config.h.windows coap_config.h COPYONLY)
-+elseif(APPLE)
-+ configure_file(include/coap2/coap.h.macos include/coap2/coap.h COPYONLY)
-+ configure_file(coap_config.h.macos coap_config.h COPYONLY)
-+endif()
-+
-+set(PUBLIC_HEADERS include/coap2/address.h
-+ include/coap2/async.h
-+ include/coap2/bits.h
-+ include/coap2/block.h
-+ ${CMAKE_BINARY_DIR}/include/coap2/coap.h
-+ include/coap2/coap_debug.h
-+ include/coap2/coap_dtls.h
-+ include/coap2/coap_event.h
-+ include/coap2/coap_forward_decls.h
-+ include/coap2/coap_hashkey.h
-+ include/coap2/coap_io.h
-+ include/coap2/coap_mutex.h
-+ include/coap2/coap_session.h
-+ include/coap2/coap_time.h
-+ include/coap2/encode.h
-+ include/coap2/libcoap.h
-+ include/coap2/mem.h
-+ include/coap2/net.h
-+ include/coap2/option.h
-+ include/coap2/pdu.h
-+ include/coap2/prng.h
-+ include/coap2/resource.h
-+ include/coap2/str.h
-+ include/coap2/subscribe.h
-+ include/coap2/uri.h
-+ include/coap2/uthash.h)
-+
-+add_library(coap STATIC ${SOURCES})
-+
-+set_property(TARGET coap PROPERTY POSITION_INDEPENDENT_CODE ON)
-+
-+target_include_directories(coap
-+ PRIVATE
-+ ${CMAKE_BINARY_DIR}
-+ PUBLIC
-+ include/coap2
-+ ${CMAKE_BINARY_DIR}/include/coap2/)
-+
-+install(TARGETS coap
-+ ARCHIVE DESTINATION lib
-+)
-+
-+install(FILES ${PUBLIC_HEADERS} DESTINATION include/coap2)
-diff -rupN orig/coap_config.h.windows patched/coap_config.h.windows
---- orig/coap_config.h.windows 2019-11-21 17:42:00.000000000 +0100
-+++ patched/coap_config.h.windows 2019-11-22 09:11:43.000000000 +0100
-@@ -67,7 +67,7 @@
- #define snprintf _snprintf
- #endif
-
--#define HAVE_OPENSSL 1
-+/* #define HAVE_OPENSSL 1 */
-
- /* Define to the address where bug reports for this package should be sent. */
- #define PACKAGE_BUGREPORT "libcoap-developers@lists.sourceforge.net"
diff --git a/win_build_vs.bat b/win_build_vs.bat
index 0cf8ea35e..02884cc5d 100644
--- a/win_build_vs.bat
+++ b/win_build_vs.bat
@@ -26,7 +26,6 @@ set skiptestrun=OFF
set cmake_build_type=Release
set build_platform=x64
set enable_kafka=ON
-set enable_coap=OFF
set enable_jni=OFF
set enable_sql=ON
set enable_aws=ON
@@ -72,7 +71,6 @@ for %%x in (%*) do (
if [%%~x] EQU [/NO_KAFKA] set enable_kafka=OFF
if [%%~x] EQU [/J] set enable_jni=ON
if [%%~x] EQU [/NO_SQL] set enable_sql=OFF
- if [%%~x] EQU [/C] set enable_coap=ON
if [%%~x] EQU [/NO_AWS] set enable_aws=OFF
if [%%~x] EQU [/SFTP] set enable_sftp=ON
if [%%~x] EQU [/PDH] set enable_pdh=ON
@@ -121,7 +119,7 @@ if [%generator%] EQU ["Ninja"] (
echo on
cmake -G %generator% %build_platform_cmd% -DMINIFI_INCLUDE_VC_REDIST_MERGE_MODULES=%installer_merge_modules% -DTEST_CUSTOM_WEL_PROVIDER=%test_custom_wel_provider% -DENABLE_SQL=%enable_sql% -DMINIFI_USE_REAL_ODBC_TEST_DRIVER=%real_odbc% ^
-DCMAKE_BUILD_TYPE_INIT=%cmake_build_type% -DCMAKE_BUILD_TYPE=%cmake_build_type% -DWIN32=WIN32 -DENABLE_LIBRDKAFKA=%enable_kafka% -DENABLE_JNI=%enable_jni% ^
- -DENABLE_COAP=%enable_coap% -DENABLE_AWS=%enable_aws% -DENABLE_PDH=%enable_pdh% -DENABLE_AZURE=%enable_azure% -DENABLE_SFTP=%enable_sftp% -DENABLE_SPLUNK=%enable_splunk% -DENABLE_GCP=%enable_gcp% ^
+ -DENABLE_AWS=%enable_aws% -DENABLE_PDH=%enable_pdh% -DENABLE_AZURE=%enable_azure% -DENABLE_SFTP=%enable_sftp% -DENABLE_SPLUNK=%enable_splunk% -DENABLE_GCP=%enable_gcp% ^
-DENABLE_NANOFI=%enable_nanofi% -DENABLE_OPENCV=%enable_opencv% -DENABLE_PROMETHEUS=%enable_prometheus% -DENABLE_ELASTICSEARCH=%enable_elastic% -DUSE_SHARED_LIBS=OFF -DENABLE_CONTROLLER=OFF ^
-DENABLE_BUSTACHE=%enable_bustache% -DENABLE_ENCRYPT_CONFIG=%enable_encrypt_config% -DENABLE_LUA_SCRIPTING=%enable_lua_scripting% -DENABLE_SMB=%enable_smb% ^
-DENABLE_MQTT=%enable_mqtt% -DENABLE_OPC=%enable_opc% -DENABLE_OPENWSMAN=%enable_openwsman% -DENABLE_OPS=%enable_ops% -DENABLE_PCAP=%enable_pcap% ^
(nifi-minifi-cpp) 06/06: MINIFICPP-2042 Deprecate the Both output format in CWEL
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit ff43ef4f402c136ec971cb1828d433c51468b8df
Author: Ferenc Gerlits <fg...@gmail.com>
AuthorDate: Mon Jun 24 10:35:17 2024 +0200
MINIFICPP-2042 Deprecate the Both output format in CWEL
Closes #1820
Signed-off-by: Martin Zink <ma...@apache.org>
---
PROCESSORS.md | 36 +++++++++++-----------
.../windows-event-log/ConsumeWindowsEventLog.h | 12 ++++----
2 files changed, 24 insertions(+), 24 deletions(-)
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 1d1f5eea7..9dd123052 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -500,29 +500,29 @@ In the list below, the names of required properties appear in bold. Any other pr
### Description
-Windows Event Log Subscribe Callback to receive FlowFiles from Events on Windows.
+Registers a Windows Event Log Subscribe Callback to receive FlowFiles from Events on Windows. These can be filtered via channel and XPath.
### Properties
In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
-| Name | Default Value | Allowable Values | Description [...]
-|------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- [...]
-| **Channel** | System | | The Windows Event Log Channel to listen to. In order to process logs from a log file use the format 'SavedLog:\<file path\>'.<br/>**Supports Expression Language: true** [...]
-| **Query** | * | | XPath Query to filter events. (See https://msdn.microsoft.com/en-us/library/windows/desktop/dd996910(v=vs.85).aspx for examples.)<br/>**Supports Expression Language: true** [...]
-| **Max Buffer Size** | 1 MB | | The individual Event Log XMLs are rendered to a buffer. This specifies the maximum size in bytes that the buffer will be allowed to grow to. (Limiting the maximum size of an individual Event XM [...]
-| **Inactive Duration To Reconnect** | 10 min | | If no new event logs are processed for the specified time period, this processor will try reconnecting to recover from a state where any further messages cannot be consumed. Such situation can [...]
-| Identifier Match Regex | .*Sid | | Regular Expression to match Subject Identifier Fields. These will be placed into the attributes of the FlowFile [...]
-| Apply Identifier Function | true | true<br/>false | If true it will resolve SIDs matched in the 'Identifier Match Regex' to the DOMAIN\USERNAME associated with that ID [...]
-| Resolve Metadata in Attributes | true | true<br/>false | If true, any metadata that is resolved ( such as IDs or keyword metadata ) will be placed into attributes, otherwise it will be replaced in the XML or text output [...]
-| Event Header Delimiter | | | If set, the chosen delimiter will be used in the Event output header. Otherwise, a colon followed by spaces will be used. [...]
-| Event Header | LOG_NAME=Log Name, SOURCE = Source, TIME_CREATED = Date,EVENT_RECORDID=Record ID,EVENTID = Event ID,TASK_CATEGORY = Task Category,LEVEL = Level,KEYWORDS = Keywords,USER = User,COMPUTER = Computer, EVENT_TYPE = EventType | | Comma seperated list of key/value pairs with the following keys LOG_NAME, SOURCE, TIME_CREATED,EVENT_RECORDID,EVENTID,TASK_CATEGORY,LEVEL,KEYWORDS,USER,COMPUTER, and EVENT_TYPE. Eliminating fie [...]
-| **Output Format** | Both | XML<br/>Both<br/>Plaintext<br/>JSON | Set the output format type. In case 'Both' is selected the processor generates two flow files for every event captured in format XML and Plaintext [...]
-| **JSON Format** | Simple | Raw<br/>Simple<br/>Flattened | Set the json format type. Only applicable if Output Format is set to 'JSON' [...]
-| Batch Commit Size | 1000 | | Maximum number of Events to consume and create to Flow Files from before committing. [...]
-| State Directory | CWELState | | DEPRECATED. Only use it for state migration from the state file, supplying the legacy state directory. [...]
-| **Process Old Events** | false | true<br/>false | This property defines if old events (which are created before first time server is started) should be processed. [...]
-| Cache SID Lookups | true | true<br/>false | Determines whether SID to name lookups are cached in memory [...]
+| Name | Default Value | Allowable Values | Description [...]
+|------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- [...]
+| **Channel** | System | | The Windows Event Log Channel to listen to. In order to process logs from a log file use the format 'SavedLog:\<file path\>'.<br/>**Supports Expression Language: true** [...]
+| **Query** | * | | XPath Query to filter events. (See https://msdn.microsoft.com/en-us/library/windows/desktop/dd996910(v=vs.85).aspx for examples.)<br/>**Supports Expression Language: true** [...]
+| **Max Buffer Size** | 1 MB | | The individual Event Log XMLs are rendered to a buffer. This specifies the maximum size in bytes that the buffer will be allowed to grow to. (Limiting the maximum size of an individual Event XML.) [...]
+| **Inactive Duration To Reconnect** | 10 min | | If no new event logs are processed for the specified time period, this processor will try reconnecting to recover from a state where any further messages cannot be consumed. Such situation can happen [...]
+| Identifier Match Regex | .*Sid | | Regular Expression to match Subject Identifier Fields. These will be placed into the attributes of the FlowFile [...]
+| Apply Identifier Function | true | true<br/>false | If true it will resolve SIDs matched in the 'Identifier Match Regex' to the DOMAIN\USERNAME associated with that ID [...]
+| Resolve Metadata in Attributes | true | true<br/>false | If true, any metadata that is resolved ( such as IDs or keyword metadata ) will be placed into attributes, otherwise it will be replaced in the XML or text output [...]
+| Event Header Delimiter | | | If set, the chosen delimiter will be used in the Event output header. Otherwise, a colon followed by spaces will be used. [...]
+| Event Header | LOG_NAME=Log Name, SOURCE = Source, TIME_CREATED = Date,EVENT_RECORDID=Record ID,EVENTID = Event ID,TASK_CATEGORY = Task Category,LEVEL = Level,KEYWORDS = Keywords,USER = User,COMPUTER = Computer, EVENT_TYPE = EventType | | Comma seperated list of key/value pairs with the following keys LOG_NAME, SOURCE, TIME_CREATED,EVENT_RECORDID,EVENTID,TASK_CATEGORY,LEVEL,KEYWORDS,USER,COMPUTER, and EVENT_TYPE. Eliminating fields wil [...]
+| **Output Format** | XML | XML<br/>Plaintext<br/>JSON | The format of the output flow files. [...]
+| **JSON Format** | Simple | Raw<br/>Simple<br/>Flattened | Set the json format type. Only applicable if Output Format is set to 'JSON' [...]
+| Batch Commit Size | 1000 | | Maximum number of Events to consume and create to Flow Files from before committing. [...]
+| State Directory | CWELState | | DEPRECATED. Only use it for state migration from the state file, supplying the legacy state directory. [...]
+| **Process Old Events** | false | true<br/>false | This property defines if old events (which are created before first time server is started) should be processed. [...]
+| Cache SID Lookups | true | true<br/>false | Determines whether SID to name lookups are cached in memory [...]
### Relationships
diff --git a/extensions/windows-event-log/ConsumeWindowsEventLog.h b/extensions/windows-event-log/ConsumeWindowsEventLog.h
index 3b0e4b187..e2509b532 100644
--- a/extensions/windows-event-log/ConsumeWindowsEventLog.h
+++ b/extensions/windows-event-log/ConsumeWindowsEventLog.h
@@ -63,7 +63,7 @@ struct EventRender {
enum class OutputFormat {
XML,
- Both,
+ Both, // Both is DEPRECATED and removed from the documentation, but kept for backwards compatibility; it means XML + Plaintext
Plaintext,
JSON
};
@@ -83,7 +83,7 @@ class ConsumeWindowsEventLog : public core::Processor {
~ConsumeWindowsEventLog() override;
- EXTENSIONAPI static constexpr const char* Description = "Windows Event Log Subscribe Callback to receive FlowFiles from Events on Windows.";
+ EXTENSIONAPI static constexpr const char* Description = "Registers a Windows Event Log Subscribe Callback to receive FlowFiles from Events on Windows. These can be filtered via channel and XPath.";
EXTENSIONAPI static constexpr auto Channel = core::PropertyDefinitionBuilder<>::createProperty("Channel")
.isRequired(true)
@@ -142,11 +142,11 @@ class ConsumeWindowsEventLog : public core::Processor {
.withDescription("Comma seperated list of key/value pairs with the following keys LOG_NAME, SOURCE, TIME_CREATED,EVENT_RECORDID,"
"EVENTID,TASK_CATEGORY,LEVEL,KEYWORDS,USER,COMPUTER, and EVENT_TYPE. Eliminating fields will remove them from the header.")
.build();
- EXTENSIONAPI static constexpr auto OutputFormatProperty = core::PropertyDefinitionBuilder<magic_enum::enum_count<cwel::OutputFormat>()>::createProperty("Output Format")
+ EXTENSIONAPI static constexpr auto OutputFormatProperty = core::PropertyDefinitionBuilder<magic_enum::enum_count<cwel::OutputFormat>() - 1>::createProperty("Output Format")
.isRequired(true)
- .withDefaultValue(magic_enum::enum_name(cwel::OutputFormat::Both))
- .withAllowedValues(magic_enum::enum_names<cwel::OutputFormat>())
- .withDescription("Set the output format type. In case \'Both\' is selected the processor generates two flow files for every event captured in format XML and Plaintext")
+ .withDefaultValue(magic_enum::enum_name(cwel::OutputFormat::XML))
+ .withAllowedValues(std::array{magic_enum::enum_name(cwel::OutputFormat::XML), magic_enum::enum_name(cwel::OutputFormat::Plaintext), magic_enum::enum_name(cwel::OutputFormat::JSON)})
+ .withDescription("The format of the output flow files.")
.build();
EXTENSIONAPI static constexpr auto JsonFormatProperty = core::PropertyDefinitionBuilder<magic_enum::enum_count<cwel::JsonFormat>()>::createProperty("JSON Format")
.isRequired(true)
(nifi-minifi-cpp) 01/06: MINIFICPP-2341 Include custom relationships in C2 manifest for NiFi python processors
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 1e9b6159be72339c4d8db341f72fcb93308b2c11
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jun 24 10:34:23 2024 +0200
MINIFICPP-2341 Include custom relationships in C2 manifest for NiFi python processors
Closes #1770
Signed-off-by: Martin Zink <ma...@apache.org>
---
docker/test/integration/cluster/ImageStore.py | 4 +-
docker/test/integration/features/python.feature | 14 ++++
...hecker.py => ProcessContextInterfaceChecker.py} | 7 +-
.../processors/SpecialPropertyTypeChecker.py | 1 +
.../python/ProcessContextInterfaceChecker.py | 92 ++++++++++++++++++++++
extensions/python/ExecutePythonProcessor.cpp | 8 ++
extensions/python/ExecutePythonProcessor.h | 2 +
extensions/python/PYTHON.md | 4 +
extensions/python/PythonCreator.h | 2 +-
extensions/python/PythonScriptEngine.cpp | 39 +++++++++
extensions/python/PythonScriptEngine.h | 31 ++++++--
.../pythonprocessors/nifiapi/flowfiletransform.py | 8 +-
.../python/pythonprocessors/nifiapi/properties.py | 49 ++++++++----
.../pythonprocessors/nifiapi/relationship.py | 15 ++--
extensions/python/types/PyProcessContext.cpp | 38 +++++++++
extensions/python/types/PyProcessContext.h | 2 +
16 files changed, 276 insertions(+), 40 deletions(-)
diff --git a/docker/test/integration/cluster/ImageStore.py b/docker/test/integration/cluster/ImageStore.py
index f8a550fcf..f82fcb792 100644
--- a/docker/test/integration/cluster/ImageStore.py
+++ b/docker/test/integration/cluster/ImageStore.py
@@ -160,6 +160,7 @@ class ImageStore:
USER minificpp
COPY RotatingForwarder.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/RotatingForwarder.py
COPY SpecialPropertyTypeChecker.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/SpecialPropertyTypeChecker.py
+ COPY ProcessContextInterfaceChecker.py /opt/minifi/minifi-current/minifi-python/nifi_python_processors/ProcessContextInterfaceChecker.py
RUN wget {parse_document_url} --directory-prefix=/opt/minifi/minifi-current/minifi-python/nifi_python_processors && \\
wget {chunk_document_url} --directory-prefix=/opt/minifi/minifi-current/minifi-python/nifi_python_processors && \\
echo 'langchain<=0.17.0' > /opt/minifi/minifi-current/minifi-python/nifi_python_processors/requirements.txt && \\
@@ -180,7 +181,8 @@ class ImageStore:
chunk_document_sed_cmd=chunk_document_sed_cmd))
return self.__build_image(dockerfile, [os.path.join(self.test_dir, "resources", "python", "RotatingForwarder.py"),
- os.path.join(self.test_dir, "resources", "python", "SpecialPropertyTypeChecker.py")])
+ os.path.join(self.test_dir, "resources", "python", "SpecialPropertyTypeChecker.py"),
+ os.path.join(self.test_dir, "resources", "python", "ProcessContextInterfaceChecker.py")])
def __build_http_proxy_image(self):
dockerfile = dedent("""\
diff --git a/docker/test/integration/features/python.feature b/docker/test/integration/features/python.feature
index b8e945905..c0761b517 100644
--- a/docker/test/integration/features/python.feature
+++ b/docker/test/integration/features/python.feature
@@ -129,3 +129,17 @@ Feature: MiNiFi can use python processors in its flows
When all instances start up
Then one flowfile with the contents "Check successful!" is placed in the monitored directory in less than 30 seconds
+
+ @USE_NIFI_PYTHON_PROCESSORS
+ Scenario: NiFi native python processor's ProcessContext interface can be used in MiNiFi C++
+ Given a GenerateFlowFile processor with the "File Size" property set to "0B"
+ And a ProcessContextInterfaceChecker processor
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And python is installed on the MiNiFi agent with a pre-created virtualenv
+
+ And the "success" relationship of the GenerateFlowFile processor is connected to the ProcessContextInterfaceChecker
+ And the "myrelationship" relationship of the ProcessContextInterfaceChecker processor is connected to the PutFile
+
+ When all instances start up
+
+ Then one flowfile with the contents "Check successful!" is placed in the monitored directory in less than 30 seconds
diff --git a/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py b/docker/test/integration/minifi/processors/ProcessContextInterfaceChecker.py
similarity index 82%
copy from docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
copy to docker/test/integration/minifi/processors/ProcessContextInterfaceChecker.py
index 58da4db54..37de61f32 100644
--- a/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
+++ b/docker/test/integration/minifi/processors/ProcessContextInterfaceChecker.py
@@ -15,11 +15,12 @@
from ..core.Processor import Processor
-class SpecialPropertyTypeChecker(Processor):
+class ProcessContextInterfaceChecker(Processor):
def __init__(self, context):
- super(SpecialPropertyTypeChecker, self).__init__(
+ super(ProcessContextInterfaceChecker, self).__init__(
context=context,
- clazz='SpecialPropertyTypeChecker',
+ clazz='ProcessContextInterfaceChecker',
class_prefix='org.apache.nifi.minifi.processors.nifi_python_processors.',
properties={},
+ schedule={'scheduling strategy': 'EVENT_DRIVEN'},
auto_terminate=[])
diff --git a/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py b/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
index 58da4db54..7125b6997 100644
--- a/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
+++ b/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
@@ -22,4 +22,5 @@ class SpecialPropertyTypeChecker(Processor):
clazz='SpecialPropertyTypeChecker',
class_prefix='org.apache.nifi.minifi.processors.nifi_python_processors.',
properties={},
+ schedule={'scheduling strategy': 'EVENT_DRIVEN'},
auto_terminate=[])
diff --git a/docker/test/integration/resources/python/ProcessContextInterfaceChecker.py b/docker/test/integration/resources/python/ProcessContextInterfaceChecker.py
new file mode 100644
index 000000000..83cf9f27a
--- /dev/null
+++ b/docker/test/integration/resources/python/ProcessContextInterfaceChecker.py
@@ -0,0 +1,92 @@
+# 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.
+
+from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult
+from nifiapi.properties import PropertyDescriptor, StandardValidators
+from nifiapi.relationship import Relationship
+
+
+class ProcessContextInterfaceChecker(FlowFileTransform):
+ """
+ Checks ProcessContext interface
+ """
+ SECRET_PASSWORD = PropertyDescriptor(
+ name="Secret Password",
+ description="Password to access",
+ validators=[StandardValidators.NON_EMPTY_VALIDATOR],
+ default_value="mysecret",
+ required=True,
+ sensitive=True
+ )
+ REQUEST_TIMEOUT = PropertyDescriptor(
+ name="Request Timeout",
+ description="Timeout for the request",
+ validators=[StandardValidators.TIME_PERIOD_VALIDATOR],
+ default_value="60 sec",
+ required=True
+ )
+ WISH_COUNT = PropertyDescriptor(
+ name="Wish Count",
+ description="Number of wishes",
+ default_value="3",
+ validators=[StandardValidators.POSITIVE_INTEGER_VALIDATOR],
+ required=True
+ )
+
+ property_descriptors = [
+ SECRET_PASSWORD,
+ REQUEST_TIMEOUT,
+ WISH_COUNT
+ ]
+
+ def __init__(self, **kwargs):
+ pass
+
+ def getPropertyDescriptors(self):
+ return self.property_descriptors
+
+ def getRelationships(self):
+ return [Relationship("myrelationship", "Dummy dynamic relationship")]
+
+ def transform(self, context, flowFile):
+ properties = context.getProperties()
+ if len(properties) != 3:
+ return FlowFileTransformResult("failure", contents="Property count is invalid")
+
+ property_names = [property.name for property in properties]
+ if "Secret Password" not in property_names or "Request Timeout" not in property_names or "Wish Count" not in property_names:
+ return FlowFileTransformResult("failure", contents="Missing properties")
+
+ for property in properties:
+ if property.name == "Secret Password" and properties[property] != "mysecret":
+ return FlowFileTransformResult("failure", contents="Secret Password value is invalid")
+ elif property.name == "Request Timeout" and properties[property] != "60 sec":
+ return FlowFileTransformResult("failure", contents="Request Timeout value is invalid")
+ elif property.name == "Wish Count" and properties[property] != "3":
+ return FlowFileTransformResult("failure", contents="Wish Count value is invalid")
+
+ secret_password = context.getProperty(self.SECRET_PASSWORD).getValue()
+ if secret_password != "mysecret":
+ return FlowFileTransformResult("failure", contents="Secret password is invalid")
+
+ timeout = context.getProperty(self.REQUEST_TIMEOUT).getValue()
+ if timeout != "60 sec":
+ return FlowFileTransformResult("failure", contents="Request timeout is invalid")
+
+ wish_count = context.getProperty(self.WISH_COUNT).getValue()
+ if wish_count != "3":
+ return FlowFileTransformResult("failure", contents="Wish count is invalid")
+
+ return FlowFileTransformResult("myrelationship", contents="Check successful!")
diff --git a/extensions/python/ExecutePythonProcessor.cpp b/extensions/python/ExecutePythonProcessor.cpp
index 087d5d43e..0d364803c 100644
--- a/extensions/python/ExecutePythonProcessor.cpp
+++ b/extensions/python/ExecutePythonProcessor.cpp
@@ -211,6 +211,14 @@ std::map<std::string, core::Property> ExecutePythonProcessor::getProperties() co
return result;
}
+std::vector<core::Relationship> ExecutePythonProcessor::getPythonRelationships() {
+ auto relationships = getSupportedRelationships();
+ auto custom_relationships = python_script_engine_->getCustomPythonRelationships();
+ relationships.reserve(relationships.size() + std::distance(custom_relationships.begin(), custom_relationships.end()));
+ relationships.insert(relationships.end(), custom_relationships.begin(), custom_relationships.end());
+ return relationships;
+}
+
REGISTER_RESOURCE(ExecutePythonProcessor, Processor);
} // namespace org::apache::nifi::minifi::extensions::python::processors
diff --git a/extensions/python/ExecutePythonProcessor.h b/extensions/python/ExecutePythonProcessor.h
index 7927fe58b..c13e85bce 100644
--- a/extensions/python/ExecutePythonProcessor.h
+++ b/extensions/python/ExecutePythonProcessor.h
@@ -125,6 +125,8 @@ class ExecutePythonProcessor : public core::Processor {
std::map<std::string, core::Property> getProperties() const override;
+ std::vector<core::Relationship> getPythonRelationships();
+
protected:
const core::Property* findProperty(const std::string& name) const override;
diff --git a/extensions/python/PYTHON.md b/extensions/python/PYTHON.md
index 97aaaecc0..ea3cfb87e 100644
--- a/extensions/python/PYTHON.md
+++ b/extensions/python/PYTHON.md
@@ -166,6 +166,10 @@ Due to some differences between the NiFi and MiNiFi C++ processors and implement
- MiNiFi C++ only supports expression language with flow file attributes, so only FLOWFILE_ATTRIBUTES expression language scope is supported, otherwise the expression language will not be evaluated.
- MiNiFi C++ does not support property dependencies, so the property dependencies will be ignored. If a property depends on another property, the property will not be required.
- MiNiFi C++ does not support the use of self.jvm member in Python processors that provides JVM bindings in NiFi, it is set to None in MiNiFi C++.
+- Dynamic properties are supported in all Python processors and the dynamic properties defined in the flow configuration are automatically added. There is no need to define getDynamicPropertyDescriptor method in the Python processor. The only caveat is that the description of the dynamic property cannot be custom defined.
+- In MiNiFi C++ when the processor is stopped the stop event handling is done in the `notifyStop()` method which does not have the context available. Due to this the `def onStopped(self, context)` cannot be called in NiFi Python processors, so the `onStopped` method is not supported in MiNiFi C++.
+- The interface of the `ProcessContext` class is a bit more limited in MiNiFi C++ compared to NiFi. The available methods in `ProcessContext` are `getProperty`, `getStateManager`, `getName` and `getProperties`.
+- Success relationship is always present in all Python processors even if custom relationships are defined in the Python processor with the `getRelationships` method.
## Use Python processors from virtualenv
diff --git a/extensions/python/PythonCreator.h b/extensions/python/PythonCreator.h
index 3dc96d0f2..15b47019f 100644
--- a/extensions/python/PythonCreator.h
+++ b/extensions/python/PythonCreator.h
@@ -121,7 +121,7 @@ class PythonCreator : public minifi::core::CoreComponent {
.full_name_ = full_name,
.description_ = processor->getDescription(),
.class_properties_ = processor->getPythonProperties(),
- .class_relationships_ = processor->getSupportedRelationships(),
+ .class_relationships_ = processor->getPythonRelationships(),
.supports_dynamic_properties_ = processor->getPythonSupportDynamicProperties(),
.inputRequirement_ = toString(processor->getInputRequirement()),
.isSingleThreaded_ = processor->isSingleThreaded()};
diff --git a/extensions/python/PythonScriptEngine.cpp b/extensions/python/PythonScriptEngine.cpp
index 568f5f9b1..d3ac5927c 100644
--- a/extensions/python/PythonScriptEngine.cpp
+++ b/extensions/python/PythonScriptEngine.cpp
@@ -204,4 +204,43 @@ void PythonScriptEngine::initializeProcessorObject(const std::string& python_cla
}
}
+std::vector<core::Relationship> PythonScriptEngine::getCustomPythonRelationships() {
+ GlobalInterpreterLock gil;
+ std::vector<core::Relationship> relationships;
+ if (processor_instance_.get() != nullptr) {
+ auto python_list_of_relationships = OwnedList(callProcessorObjectMethod("getRelationships"));
+ if (python_list_of_relationships.get() == Py_None) {
+ return relationships;
+ }
+ try {
+ for (size_t i = 0; i < python_list_of_relationships.length(); ++i) {
+ if (PyObject_HasAttrString(python_list_of_relationships[i].get(), "name") == 0) {
+ logger_->log_error("Error in python processor getRelationships method: Custom relationship object has no name attribute!");
+ continue;
+ }
+ auto name = OwnedStr(PyObject_GetAttrString(python_list_of_relationships[i].get(), "name"));
+ if (name.get() == nullptr) {
+ logger_->log_error("Error in python processor getRelationships method: Custom relationship object's name attribute could not be read!");
+ continue;
+ }
+
+ if (PyObject_HasAttrString(python_list_of_relationships[i].get(), "description") == 0) {
+ logger_->log_error("Error in python processor getRelationships method: Custom relationship object has no description attribute!");
+ continue;
+ }
+ auto description = OwnedStr(PyObject_GetAttrString(python_list_of_relationships[i].get(), "description"));
+ if (description.get() == nullptr) {
+ logger_->log_error("Error in python processor getRelationships method: Custom relationship object's description attribute could not be read!");
+ continue;
+ }
+
+ relationships.push_back(core::Relationship(name.toUtf8String(), description.toUtf8String()));
+ }
+ } catch (const std::exception& e) {
+ throw PythonScriptException(e.what());
+ }
+ }
+ return relationships;
+}
+
} // namespace org::apache::nifi::minifi::extensions::python
diff --git a/extensions/python/PythonScriptEngine.h b/extensions/python/PythonScriptEngine.h
index a25edf409..401ee0fa3 100644
--- a/extensions/python/PythonScriptEngine.h
+++ b/extensions/python/PythonScriptEngine.h
@@ -35,6 +35,8 @@
#include "PythonScriptException.h"
#include "properties/Configuration.h"
#include "PythonInterpreter.h"
+#include "core/logging/Logger.h"
+#include "core/logging/LoggerConfiguration.h"
namespace org::apache::nifi::minifi::extensions::python {
@@ -56,7 +58,7 @@ class PythonScriptEngine {
}
template<typename... Args>
- void call(const char* fn_name, Args&& ...args) {
+ OwnedReference call(const char* fn_name, Args&& ...args) {
GlobalInterpreterLock gil_lock;
try {
if (auto item = bindings_[fn_name]) {
@@ -64,63 +66,76 @@ class PythonScriptEngine {
if (!result) {
throw PyException();
}
+ return result;
}
+ return OwnedReference(Py_None);
} catch (const std::exception& e) {
throw PythonScriptException(e.what());
}
}
template<typename ... Args>
- void callRequiredFunction(const std::string& fn_name, Args&& ...args) {
+ OwnedReference callRequiredFunction(const std::string& fn_name, Args&& ...args) {
GlobalInterpreterLock gil_lock;
if (auto item = bindings_[fn_name]) {
auto result = BorrowedCallable(*item)(std::forward<Args>(args)...);
if (!result) {
throw PyException();
}
+ return result;
} else {
throw std::runtime_error("Required Function '" + fn_name + "' is not found within Python bindings");
}
}
template<typename ... Args>
- void callProcessorObjectMethod(const std::string& fn_name, Args&& ...args) {
+ OwnedReference callProcessorObjectMethod(const std::string& fn_name, Args&& ...args) {
GlobalInterpreterLock gil_lock;
if (processor_instance_.get() == nullptr) {
throw std::runtime_error("No python processor instance is set!");
}
try {
+ if (PyObject_HasAttrString(processor_instance_.get(), fn_name.c_str()) == 0) {
+ return OwnedReference(Py_None);
+ }
+
auto callable_method = OwnedCallable(PyObject_GetAttrString(processor_instance_.get(), fn_name.c_str()));
if (callable_method.get() == nullptr) {
- return;
+ return OwnedReference(Py_None);
}
auto result = callable_method(std::forward<Args>(args)...);
- if (!result) {
+ if (!result.get()) {
throw PyException();
}
+ return result;
} catch (const std::exception& e) {
throw PythonScriptException(e.what());
}
}
template<typename ... Args>
- void callRequiredProcessorObjectMethod(const std::string& fn_name, Args&& ...args) {
+ OwnedReference callRequiredProcessorObjectMethod(const std::string& fn_name, Args&& ...args) {
GlobalInterpreterLock gil_lock;
if (processor_instance_.get() == nullptr) {
throw std::runtime_error("No python processor instance is set!");
}
+ if (PyObject_HasAttrString(processor_instance_.get(), fn_name.c_str()) == 0) {
+ throw std::runtime_error("Required method '" + fn_name + "' is not found in python processor object!");
+ }
+
auto callable_method = OwnedCallable(PyObject_GetAttrString(processor_instance_.get(), fn_name.c_str()));
if (callable_method.get() == nullptr) {
- throw std::runtime_error("Required method '" + fn_name + "' is not found in python processor object!");
+ throw std::runtime_error("Failed to get required method '" + fn_name + "' is in python processor object!");
}
auto result = callable_method(std::forward<Args>(args)...);
if (!result) {
throw PyException();
}
+ return result;
}
template<object::convertible T>
@@ -135,6 +150,7 @@ class PythonScriptEngine {
void onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session);
void initialize(const core::Relationship& success, const core::Relationship& failure, const core::Relationship& original, const std::shared_ptr<core::logging::Logger>& logger);
void initializeProcessorObject(const std::string& python_class_name);
+ std::vector<core::Relationship> getCustomPythonRelationships();
private:
void evalInternal(std::string_view script);
@@ -144,6 +160,7 @@ class PythonScriptEngine {
OwnedObject processor_instance_;
std::optional<std::string> processor_class_name_;
std::vector<std::filesystem::path> module_paths_;
+ std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<PythonScriptEngine>::getLogger();
};
} // namespace org::apache::nifi::minifi::extensions::python
diff --git a/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py b/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py
index 0420b90f9..d17facd05 100644
--- a/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py
+++ b/extensions/python/pythonprocessors/nifiapi/flowfiletransform.py
@@ -16,7 +16,9 @@
import traceback
from abc import ABC, abstractmethod
from typing import List
-from .properties import ExpressionLanguageScope, FlowFileProxy, ProcessContextProxy, PropertyDescriptor, translateStandardValidatorToMiNiFiPropertype
+from .properties import ExpressionLanguageScope, PropertyDescriptor, translateStandardValidatorToMiNiFiPropertype
+from .properties import FlowFile as FlowFileProxy
+from .properties import ProcessContext as ProcessContextProxy
from minifi_native import OutputStream, Processor, ProcessContext, ProcessSession
@@ -76,7 +78,7 @@ class FlowFileTransform(ABC):
pass
def onSchedule(self, context: ProcessContext):
- context_proxy = ProcessContextProxy(context)
+ context_proxy = ProcessContextProxy(context, self)
self.onScheduled(context_proxy)
def onTrigger(self, context: ProcessContext, session: ProcessSession):
@@ -87,7 +89,7 @@ class FlowFileTransform(ABC):
flow_file = session.clone(original_flow_file)
flow_file_proxy = FlowFileProxy(session, flow_file)
- context_proxy = ProcessContextProxy(context)
+ context_proxy = ProcessContextProxy(context, self)
try:
result = self.transform(context_proxy, flow_file_proxy)
except Exception:
diff --git a/extensions/python/pythonprocessors/nifiapi/properties.py b/extensions/python/pythonprocessors/nifiapi/properties.py
index 5968774bf..20bf34bb3 100644
--- a/extensions/python/pythonprocessors/nifiapi/properties.py
+++ b/extensions/python/pythonprocessors/nifiapi/properties.py
@@ -14,8 +14,10 @@
# limitations under the License.
from enum import Enum
-from typing import List
-from minifi_native import ProcessSession, FlowFile, ProcessContext, timePeriodStringToMilliseconds, dataSizeStringToBytes
+from typing import List, Dict
+from minifi_native import ProcessSession, StateManager, timePeriodStringToMilliseconds, dataSizeStringToBytes
+from minifi_native import FlowFile as CppFlowFile
+from minifi_native import ProcessContext as CppProcessContext
# This is a mock for NiFi's StandardValidators class methods, that return the property type equivalent in MiNiFi C++ if exists
@@ -173,26 +175,26 @@ class DataUnit(Enum):
TB = "TB"
-class FlowFileProxy:
- def __init__(self, session: ProcessSession, flow_file: FlowFile):
+class FlowFile:
+ def __init__(self, session: ProcessSession, cpp_flow_file: CppFlowFile):
self.session = session
- self.flow_file = flow_file
+ self.cpp_flow_file = cpp_flow_file
def getContentsAsBytes(self):
- return self.session.getContentsAsBytes(self.flow_file)
+ return self.session.getContentsAsBytes(self.cpp_flow_file)
def getAttribute(self, name: str):
- return self.flow_file.getAttribute(name)
+ return self.cpp_flow_file.getAttribute(name)
def getSize(self):
- return self.flow_file.getSize()
+ return self.cpp_flow_file.getSize()
def getAttributes(self):
- return self.flow_file.getAttributes()
+ return self.cpp_flow_file.getAttributes()
class PythonPropertyValue:
- def __init__(self, cpp_context: ProcessContext, name: str, string_value: str, el_supported: bool, controller_service_definition: str):
+ def __init__(self, cpp_context: CppProcessContext, name: str, string_value: str, el_supported: bool, controller_service_definition: str):
self.cpp_context = cpp_context
self.value = None
self.name = name
@@ -258,12 +260,12 @@ class PythonPropertyValue:
return float(bytes / 1024 / 1024 / 1024 / 1024)
return 0
- def evaluateAttributeExpressions(self, flow_file_proxy: FlowFileProxy = None):
- if flow_file_proxy is None or not self.el_supported:
+ def evaluateAttributeExpressions(self, flow_file: FlowFile = None):
+ if flow_file is None or not self.el_supported:
return self
# If Expression Language is supported and present, evaluate it and return a new PropertyValue.
# Otherwise just return self, in order to avoid the cost of making the call to cpp for getProperty
- new_string_value = self.cpp_context.getProperty(self.name, flow_file_proxy.flow_file)
+ new_string_value = self.cpp_context.getProperty(self.name, flow_file.cpp_flow_file)
return PythonPropertyValue(self.cpp_context, self.name, new_string_value, self.el_supported, self.controller_service_definition)
def asControllerService(self):
@@ -272,9 +274,10 @@ class PythonPropertyValue:
return self.cpp_context.getControllerService(self.value, self.controller_service_definition)
-class ProcessContextProxy:
- def __init__(self, cpp_context: ProcessContext):
+class ProcessContext:
+ def __init__(self, cpp_context: CppProcessContext, processor):
self.cpp_context = cpp_context
+ self.processor = processor
def getProperty(self, descriptor) -> PythonPropertyValue:
if descriptor is None:
@@ -289,3 +292,19 @@ class ProcessContextProxy:
controller_service_definition = descriptor.controllerServiceDefinition
property_value = self.cpp_context.getProperty(property_name)
return PythonPropertyValue(self.cpp_context, property_name, property_value, expression_language_support, controller_service_definition)
+
+ def getStateManager(self) -> StateManager:
+ return self.cpp_context.getStateManager()
+
+ def getName(self) -> str:
+ return self.cpp_context.getName()
+
+ def getProperties(self) -> Dict[PropertyDescriptor, str]:
+ properties = dict()
+ cpp_properties = self.cpp_context.getProperties()
+
+ for property_descriptor in self.processor.getPropertyDescriptors():
+ if property_descriptor.name in cpp_properties:
+ properties[property_descriptor] = cpp_properties[property_descriptor.name]
+
+ return properties
diff --git a/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py b/extensions/python/pythonprocessors/nifiapi/relationship.py
similarity index 66%
copy from docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
copy to extensions/python/pythonprocessors/nifiapi/relationship.py
index 58da4db54..2504c5d80 100644
--- a/docker/test/integration/minifi/processors/SpecialPropertyTypeChecker.py
+++ b/extensions/python/pythonprocessors/nifiapi/relationship.py
@@ -12,14 +12,9 @@
# 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.
-from ..core.Processor import Processor
-
-class SpecialPropertyTypeChecker(Processor):
- def __init__(self, context):
- super(SpecialPropertyTypeChecker, self).__init__(
- context=context,
- clazz='SpecialPropertyTypeChecker',
- class_prefix='org.apache.nifi.minifi.processors.nifi_python_processors.',
- properties={},
- auto_terminate=[])
+class Relationship:
+ def __init__(self, name, description, auto_terminated=False):
+ self.name = name
+ self.description = description
+ self.auto_terminated = auto_terminated
diff --git a/extensions/python/types/PyProcessContext.cpp b/extensions/python/types/PyProcessContext.cpp
index 31920baf9..540a74c48 100644
--- a/extensions/python/types/PyProcessContext.cpp
+++ b/extensions/python/types/PyProcessContext.cpp
@@ -21,6 +21,7 @@ a * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#include "PyStateManager.h"
#include "PyScriptFlowFile.h"
+#include "core/Processor.h"
extern "C" {
namespace org::apache::nifi::minifi::extensions::python {
@@ -29,6 +30,8 @@ static PyMethodDef PyProcessContext_methods[] = { // NOLINT(cppcoreguidelines-a
{"getProperty", (PyCFunction) PyProcessContext::getProperty, METH_VARARGS, nullptr},
{"getStateManager", (PyCFunction) PyProcessContext::getStateManager, METH_VARARGS, nullptr},
{"getControllerService", (PyCFunction) PyProcessContext::getControllerService, METH_VARARGS, nullptr},
+ {"getName", (PyCFunction) PyProcessContext::getName, METH_VARARGS, nullptr},
+ {"getProperties", (PyCFunction) PyProcessContext::getProperties, METH_VARARGS, nullptr},
{} /* Sentinel */
};
@@ -130,6 +133,41 @@ PyObject* PyProcessContext::getControllerService(PyProcessContext* self, PyObjec
Py_RETURN_NONE;
}
+PyObject* PyProcessContext::getName(PyProcessContext* self, PyObject*) {
+ auto context = self->process_context_.lock();
+ if (!context) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading process context outside 'on_trigger'");
+ return nullptr;
+ }
+
+ return object::returnReference(context->getProcessorNode()->getName());
+}
+
+PyObject* PyProcessContext::getProperties(PyProcessContext* self, PyObject*) {
+ auto context = self->process_context_.lock();
+ if (!context) {
+ PyErr_SetString(PyExc_AttributeError, "tried reading process context outside 'on_trigger'");
+ return nullptr;
+ }
+
+ auto processor = dynamic_cast<core::Processor*>(context->getProcessorNode()->getProcessor());
+ if (!processor) {
+ PyErr_SetString(PyExc_AttributeError, "Processor not available in getProperties");
+ return nullptr;
+ }
+ auto properties = processor->getProperties();
+ auto py_properties = OwnedDict::create();
+ for (const auto& [property_name, property] : properties) {
+ std::string value;
+ if (!context->getProperty(property_name, value)) {
+ continue;
+ }
+ py_properties.put(property_name, value);
+ }
+
+ return object::returnReference(py_properties);
+}
+
PyTypeObject* PyProcessContext::typeObject() {
static OwnedObject PyProcessContextType{PyType_FromSpec(&PyProcessContextTypeSpec)};
return reinterpret_cast<PyTypeObject*>(PyProcessContextType.get());
diff --git a/extensions/python/types/PyProcessContext.h b/extensions/python/types/PyProcessContext.h
index b484344b6..26aba677d 100644
--- a/extensions/python/types/PyProcessContext.h
+++ b/extensions/python/types/PyProcessContext.h
@@ -37,6 +37,8 @@ struct PyProcessContext {
static PyObject* getProperty(PyProcessContext* self, PyObject* args);
static PyObject* getStateManager(PyProcessContext* self, PyObject* args);
static PyObject* getControllerService(PyProcessContext* self, PyObject* args);
+ static PyObject* getName(PyProcessContext* self, PyObject* args);
+ static PyObject* getProperties(PyProcessContext* self, PyObject* args);
static PyTypeObject* typeObject();
};
(nifi-minifi-cpp) 05/06: MINIFICPP-2406 Fix formatting of parameters in some remaining log lines
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 7585d5d2158ff41a72807207f1d3a9876dcca082
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jun 24 10:35:08 2024 +0200
MINIFICPP-2406 Fix formatting of parameters in some remaining log lines
Closes #1818
Signed-off-by: Martin Zink <ma...@apache.org>
---
extensions/rocksdb-repos/FlowFileRepository.cpp | 4 ++--
extensions/standard-processors/processors/JoltTransformJSON.cpp | 2 +-
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/extensions/rocksdb-repos/FlowFileRepository.cpp b/extensions/rocksdb-repos/FlowFileRepository.cpp
index abdc2fcf0..e6ba15155 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.cpp
+++ b/extensions/rocksdb-repos/FlowFileRepository.cpp
@@ -158,7 +158,7 @@ void FlowFileRepository::initialize_repository() {
}
const auto container = getContainer(container_id.to_string());
if (!container) {
- logger_->log_warn("Could not find connection for %s, path %s ", container_id.to_string(), eventRead->getContentFullPath());
+ logger_->log_warn("Could not find connection for {}, path {}", container_id.to_string(), eventRead->getContentFullPath());
keys_to_delete_.enqueue({.key = key, .content = eventRead->getResourceClaim()});
continue;
}
@@ -168,7 +168,7 @@ void FlowFileRepository::initialize_repository() {
continue;
}
- logger_->log_debug("Found connection for %s, path %s ", container_id.to_string(), eventRead->getContentFullPath());
+ logger_->log_debug("Found connection for {}, path {}", container_id.to_string(), eventRead->getContentFullPath());
eventRead->setStoredToRepository(true);
// we found the connection for the persistent flowFile
// even if a processor immediately marks it for deletion, flush only happens after prune_stored_flowfiles
diff --git a/extensions/standard-processors/processors/JoltTransformJSON.cpp b/extensions/standard-processors/processors/JoltTransformJSON.cpp
index e7319399f..55a30bfa0 100644
--- a/extensions/standard-processors/processors/JoltTransformJSON.cpp
+++ b/extensions/standard-processors/processors/JoltTransformJSON.cpp
@@ -63,7 +63,7 @@ void JoltTransformJSON::onTrigger(core::ProcessContext& context, core::ProcessSe
session.writeBuffer(flowfile, std::span<const char>(buffer.GetString(), buffer.GetSize()));
session.transfer(flowfile, Success);
} else {
- logger_->log_info("Failed to apply transformation: %s", result.error());
+ logger_->log_info("Failed to apply transformation: {}", result.error());
session.transfer(flowfile, Failure);
}
}