You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2022/01/04 15:28:20 UTC
[nifi-minifi-cpp] branch main updated (e843450 -> 4f0dcd8)
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git.
from e843450 MINIFICPP-1704 Update version number to 0.12.0
new 24d9f3a MINIFICPP-1698 Make archive read/write agent-wide available
new c4769dd MINIFICPP-1290 Create test coverage for OPC processors
new 2231a72 MINIFICPP-1555 Improve the coverage of GetFileTests
new 85c4f1c MINIFICPP-1673 ProcessSession write/append should only operate on new/modified flowfiles
new 4f0dcd8 MINIFICPP-1222 Remove engine queue from ExecutePythonProcessor
The 5 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 | 2 +-
PROCESSORS.md | 4 +-
cmake/BundledOpen62541.cmake | 4 +-
.../integration/MiNiFi_integration_test_driver.py | 21 +-
docker/test/integration/features/opcua.feature | 155 ++++++++++++
.../minifi/core/AzureStorageServerContainer.py | 7 +-
docker/test/integration/minifi/core/Container.py | 3 +-
.../minifi/core/DockerTestDirectoryBindings.py | 1 +
.../integration/minifi/core/FileSystemObserver.py | 37 +--
.../test/integration/minifi/core/FlowContainer.py | 4 +-
.../integration/minifi/core/HttpProxyContainer.py | 7 +-
.../minifi/core/KafkaBrokerContainer.py | 7 +-
.../integration/minifi/core/MinifiContainer.py | 8 +-
.../integration/minifi/core/MqttBrokerContainer.py | 7 +-
.../test/integration/minifi/core/NifiContainer.py | 13 +-
.../minifi/core/OPCUAServerContainer.py | 24 ++
.../minifi/core/PostgreSQLServerContainer.py | 7 +-
.../integration/minifi/core/S3ServerContainer.py | 7 +-
.../minifi/core/SingleNodeDockerCluster.py | 23 +-
.../integration/minifi/core/ZookeeperContainer.py | 7 +-
.../minifi/processors/FetchOPCProcessor.py | 8 +
.../minifi/processors/PutOPCProcessor.py | 9 +
.../resources/opcua/opcua_client_cert.der | Bin 0 -> 1018 bytes
.../resources/opcua/opcua_client_key.der | Bin 0 -> 1192 bytes
docker/test/integration/steps/steps.py | 38 ++-
.../http-curl/tests/unit/InvokeHTTPTests.cpp | 20 --
extensions/libarchive/ArchiveStreamProvider.cpp | 49 ++++
extensions/libarchive/CompressContent.cpp | 77 ++++--
extensions/libarchive/CompressContent.h | 276 +--------------------
extensions/libarchive/ReadArchiveStream.cpp | 88 +++++++
extensions/libarchive/ReadArchiveStream.h | 89 +++++++
extensions/libarchive/WriteArchiveStream.cpp | 131 ++++++++++
extensions/libarchive/WriteArchiveStream.h | 86 +++++++
extensions/opc/include/fetchopc.h | 4 +
extensions/opc/include/opc.h | 4 +-
extensions/opc/include/putopc.h | 4 +
extensions/opc/src/opc.cpp | 68 ++---
extensions/opc/src/opcbase.cpp | 61 +++--
extensions/opc/src/putopc.cpp | 16 +-
.../script/python/ExecutePythonProcessor.cpp | 75 ++----
extensions/script/python/ExecutePythonProcessor.h | 30 +--
.../tests/unit/GetFileTests.cpp | 211 +++++++++++-----
libminifi/include/core/ProcessSession.h | 8 +
.../FileReaderCallback.h => io/ArchiveStream.h} | 47 ++--
libminifi/include/io/StreamPipe.h | 28 ++-
libminifi/src/core/ProcessSession.cpp | 10 +
.../test/archive-tests/ArchiveStreamTests.cpp | 55 ++++
thirdparty/open62541/open62541.patch | 50 ++--
48 files changed, 1230 insertions(+), 660 deletions(-)
create mode 100644 docker/test/integration/features/opcua.feature
create mode 100644 docker/test/integration/minifi/core/OPCUAServerContainer.py
create mode 100644 docker/test/integration/minifi/processors/FetchOPCProcessor.py
create mode 100644 docker/test/integration/minifi/processors/PutOPCProcessor.py
create mode 100644 docker/test/integration/resources/opcua/opcua_client_cert.der
create mode 100644 docker/test/integration/resources/opcua/opcua_client_key.der
create mode 100644 extensions/libarchive/ArchiveStreamProvider.cpp
create mode 100644 extensions/libarchive/ReadArchiveStream.cpp
create mode 100644 extensions/libarchive/ReadArchiveStream.h
create mode 100644 extensions/libarchive/WriteArchiveStream.cpp
create mode 100644 extensions/libarchive/WriteArchiveStream.h
copy libminifi/include/{utils/FileReaderCallback.h => io/ArchiveStream.h} (50%)
create mode 100644 libminifi/test/archive-tests/ArchiveStreamTests.cpp
[nifi-minifi-cpp] 03/05: MINIFICPP-1555 Improve the coverage of GetFileTests
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 2231a729633952699a23b31bbab16a692b63c3c9
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Jan 4 16:18:51 2022 +0100
MINIFICPP-1555 Improve the coverage of GetFileTests
Closes #1217
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.../tests/unit/GetFileTests.cpp | 211 +++++++++++++++------
1 file changed, 149 insertions(+), 62 deletions(-)
diff --git a/extensions/standard-processors/tests/unit/GetFileTests.cpp b/extensions/standard-processors/tests/unit/GetFileTests.cpp
index c20a739..6929dd2 100644
--- a/extensions/standard-processors/tests/unit/GetFileTests.cpp
+++ b/extensions/standard-processors/tests/unit/GetFileTests.cpp
@@ -21,7 +21,6 @@
#include <fstream>
#include "TestBase.h"
-#include "TestUtils.h"
#include "LogAttribute.h"
#include "GetFile.h"
#include "utils/file/FileUtils.h"
@@ -32,60 +31,63 @@
#include <fileapi.h>
#endif
+using namespace std::chrono_literals; // NOLINT using namespace directive is required for literals
+
namespace {
class GetFileTestController {
public:
GetFileTestController();
+ std::string getFullPath(const std::string filename) const;
+ std::string getInputFilePath() const;
void setProperty(const core::Property& property, const std::string& value);
void runSession();
TestController test_controller_;
std::shared_ptr<TestPlan> test_plan_;
- std::shared_ptr<core::Processor> get_file_processor_;
+ std::string temp_dir_;
std::string input_file_name_;
+ std::string large_input_file_name_;
+ std::string hidden_input_file_name_;
+ std::shared_ptr<core::Processor> get_file_processor_;
};
-GetFileTestController::GetFileTestController() {
+GetFileTestController::GetFileTestController()
+ : test_plan_(test_controller_.createPlan()),
+ temp_dir_(test_controller_.createTempDirectory()),
+ input_file_name_("test.txt"),
+ large_input_file_name_("large_file.txt"),
+ hidden_input_file_name_(".test.txt") {
LogTestController::getInstance().setTrace<TestPlan>();
LogTestController::getInstance().setTrace<minifi::processors::GetFile>();
LogTestController::getInstance().setTrace<minifi::processors::LogAttribute>();
- test_plan_ = test_controller_.createPlan();
- auto repo = std::make_shared<TestRepository>();
-
- auto temp_dir = test_controller_.createTempDirectory();
- REQUIRE(!temp_dir.empty());
-
- // Define test input file
- input_file_name_ = temp_dir + utils::file::FileUtils::get_separator() + "test.txt";
- std::string large_input_file_name = temp_dir + utils::file::FileUtils::get_separator() + "large_test_file.txt";
- std::string hidden_input_file_name = temp_dir + utils::file::FileUtils::get_separator() + ".test.txt";
+ REQUIRE(!temp_dir_.empty());
// Build MiNiFi processing graph
get_file_processor_ = test_plan_->addProcessor("GetFile", "Get");
- test_plan_->setProperty(get_file_processor_, minifi::processors::GetFile::Directory.getName(), temp_dir);
+ test_plan_->setProperty(get_file_processor_, minifi::processors::GetFile::Directory.getName(), temp_dir_);
auto log_attr = test_plan_->addProcessor("LogAttribute", "Log", core::Relationship("success", "description"), true);
test_plan_->setProperty(log_attr, minifi::processors::LogAttribute::FlowFilesToLog.getName(), "0");
- // Write test input.
- std::ofstream in_file_stream(input_file_name_);
- in_file_stream << "The quick brown fox jumps over the lazy dog" << std::endl;
- in_file_stream.close();
-
- in_file_stream.open(large_input_file_name);
- in_file_stream << "The quick brown fox jumps over the lazy dog who is 2 legit to quit" << std::endl;
- in_file_stream.close();
+ utils::putFileToDir(temp_dir_, input_file_name_, "The quick brown fox jumps over the lazy dog\n");
+ utils::putFileToDir(temp_dir_, large_input_file_name_, "The quick brown fox jumps over the lazy dog who is 2 legit to quit\n");
+ utils::putFileToDir(temp_dir_, hidden_input_file_name_, "But noone has ever seen it\n");
- std::ofstream hidden_in_file_stream(hidden_input_file_name);
- hidden_in_file_stream << "But noone has ever seen it" << std::endl;
- hidden_in_file_stream.close();
#ifdef WIN32
- const auto hide_file_err = utils::file::FileUtils::hide_file(hidden_input_file_name.c_str());
+ const auto hide_file_err = utils::file::FileUtils::hide_file(getFullPath(hidden_input_file_name_).c_str());
REQUIRE(!hide_file_err);
#endif
}
+std::string GetFileTestController::getFullPath(const std::string filename) const {
+ return temp_dir_ + utils::file::FileUtils::get_separator() + filename;
+}
+
+std::string GetFileTestController::getInputFilePath() const {
+ return getFullPath(input_file_name_);
+}
+
void GetFileTestController::setProperty(const core::Property& property, const std::string& value) {
test_plan_->setProperty(get_file_processor_, property.getName(), value);
}
@@ -105,23 +107,27 @@ TEST_CASE("GetFile ignores hidden files and files larger than MaxSize", "[GetFil
test_controller.runSession();
REQUIRE(LogTestController::getInstance().contains("Logged 1 flow files")); // The hidden and the too big files should be ignored
- // Check log output on windows std::endl; will produce \r\n can write manually but might as well just
- // account for the size difference here
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:test.txt"));
REQUIRE(LogTestController::getInstance().contains("key:flow.id"));
-#ifdef WIN32
- REQUIRE(LogTestController::getInstance().contains("Size:45 Offset:0"));
-#else
REQUIRE(LogTestController::getInstance().contains("Size:44 Offset:0"));
-#endif
+}
+
+TEST_CASE("GetFile ignores files smaller than MinSize", "[GetFile]") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::MinSize, "50 B");
+
+ test_controller.runSession();
+
+ REQUIRE(LogTestController::getInstance().contains("Logged 1 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:large_file.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:flow.id"));
+ REQUIRE(LogTestController::getInstance().contains("Size:67 Offset:0"));
}
TEST_CASE("GetFile onSchedule() throws if the required Directory property is not set", "[GetFile]") {
- TestController test_controller;
- LogTestController::getInstance().setTrace<TestPlan>();
- LogTestController::getInstance().setTrace<minifi::processors::GetFile>();
- auto plan = test_controller.createPlan();
- auto get_file = plan->addProcessor("GetFile", "Get");
- REQUIRE_THROWS_AS(plan->runNextProcessor(), minifi::Exception);
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::Directory, "");
+ REQUIRE_THROWS_AS(test_controller.test_plan_->runNextProcessor(), minifi::Exception);
}
TEST_CASE("GetFile removes the source file if KeepSourceFile is false") {
@@ -131,7 +137,7 @@ TEST_CASE("GetFile removes the source file if KeepSourceFile is false") {
test_controller.runSession();
- REQUIRE_FALSE(utils::file::FileUtils::exists(test_controller.input_file_name_));
+ REQUIRE_FALSE(utils::file::FileUtils::exists(test_controller.getInputFilePath()));
}
TEST_CASE("GetFile keeps the source file if KeepSourceFile is true") {
@@ -140,37 +146,118 @@ TEST_CASE("GetFile keeps the source file if KeepSourceFile is true") {
test_controller.runSession();
- REQUIRE(utils::file::FileUtils::exists(test_controller.input_file_name_));
+ REQUIRE(utils::file::FileUtils::exists(test_controller.getInputFilePath()));
}
-TEST_CASE("GetFileHiddenPropertyCheck", "[getFileProperty]") {
- TestController testController;
- LogTestController::getInstance().setTrace<TestPlan>();
- LogTestController::getInstance().setTrace<minifi::processors::GetFile>();
- LogTestController::getInstance().setTrace<minifi::processors::LogAttribute>();
- auto plan = testController.createPlan();
+TEST_CASE("Hidden files are read when IgnoreHiddenFile property is false", "[getFileProperty]") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::IgnoreHiddenFile, "false");
- auto temp_path = testController.createTempDirectory();
- std::string in_file(temp_path + utils::file::FileUtils::get_separator() + "testfifo");
- std::string hidden_in_file(temp_path + utils::file::FileUtils::get_separator() + ".testfifo");
+ test_controller.runSession();
- auto get_file = plan->addProcessor("GetFile", "Get");
- plan->setProperty(get_file, minifi::processors::GetFile::IgnoreHiddenFile.getName(), "false");
+ REQUIRE(LogTestController::getInstance().contains("Logged 3 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:large_file.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:test.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:.test.txt"));
+}
- plan->setProperty(get_file, minifi::processors::GetFile::Directory.getName(), temp_path);
- auto log_attr = plan->addProcessor("LogAttribute", "Log", core::Relationship("success", "description"), true);
- plan->setProperty(log_attr, minifi::processors::LogAttribute::FlowFilesToLog.getName(), "0");
+TEST_CASE("Check if subdirectories are ignored or not if Recurse property is set", "[getFileProperty]") {
+ GetFileTestController test_controller;
+
+ auto subdir_path = test_controller.getFullPath("subdir");
+ utils::file::FileUtils::create_dir(subdir_path);
+ utils::putFileToDir(subdir_path, "subfile.txt", "Some content in a subfile\n");
- std::ofstream in_file_stream(in_file);
- in_file_stream << "This file is not hidden" << std::endl;
- in_file_stream.close();
+ SECTION("File in subdirectory is ignored when Recurse property set to false") {
+ test_controller.setProperty(minifi::processors::GetFile::Recurse, "false");
+ test_controller.runSession();
- std::ofstream hidden_in_file_stream(hidden_in_file);
- hidden_in_file_stream << "This file is hidden" << std::endl;
- hidden_in_file_stream.close();
+ REQUIRE(LogTestController::getInstance().contains("Logged 2 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:test.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:large_file.txt"));
+ }
- plan->runNextProcessor();
- plan->runNextProcessor();
+ SECTION("File in subdirectory is logged when Recurse property set to true") {
+ test_controller.setProperty(minifi::processors::GetFile::Recurse, "true");
+ test_controller.runSession();
+
+ REQUIRE(LogTestController::getInstance().contains("Logged 3 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:test.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:large_file.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:subfile.txt"));
+ }
+}
+
+TEST_CASE("Only older files are read when MinAge property is set", "[getFileProperty]") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::MinAge, "1 hour");
+
+ const auto more_than_an_hour_ago = std::chrono::system_clock::now() - 65min;
+ utils::file::FileUtils::set_last_write_time(test_controller.getInputFilePath(),
+ std::chrono::duration_cast<std::chrono::seconds>(more_than_an_hour_ago.time_since_epoch()).count());
+
+ test_controller.runSession();
+
+ REQUIRE(LogTestController::getInstance().contains("Logged 1 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:test.txt"));
+ REQUIRE(LogTestController::getInstance().contains("Size:44 Offset:0"));
+}
+
+TEST_CASE("Only newer files are read when MaxAge property is set", "[getFileProperty]") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::MaxAge, "1 hour");
+
+ const auto more_than_an_hour_ago = std::chrono::system_clock::now() - 65min;
+ utils::file::FileUtils::set_last_write_time(test_controller.getInputFilePath(),
+ std::chrono::duration_cast<std::chrono::seconds>(more_than_an_hour_ago.time_since_epoch()).count());
+
+ test_controller.runSession();
+
+ REQUIRE(LogTestController::getInstance().contains("Logged 1 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:large_file.txt"));
+ REQUIRE(LogTestController::getInstance().contains("Size:67 Offset:0"));
+}
+
+TEST_CASE("Test BatchSize property for the maximum number of files read at once", "[getFileProperty]") {
+ GetFileTestController test_controller;
+
+ SECTION("BatchSize is set to 1 so only 1 file should be logged") {
+ test_controller.setProperty(minifi::processors::GetFile::BatchSize, "1");
+ test_controller.runSession();
+ REQUIRE(LogTestController::getInstance().contains("Logged 1 flow files"));
+ }
+
+ SECTION("BatchSize is set to 5 so all 2 non-hidden files should be logged") {
+ test_controller.setProperty(minifi::processors::GetFile::BatchSize, "5");
+ test_controller.runSession();
+ REQUIRE(LogTestController::getInstance().contains("Logged 2 flow files"));
+ }
+}
+
+TEST_CASE("Test file filtering of GetFile", "[getFileProperty]") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::FileFilter, ".?test\\.txt$");
+ test_controller.setProperty(minifi::processors::GetFile::IgnoreHiddenFile, "false");
+
+ test_controller.runSession();
REQUIRE(LogTestController::getInstance().contains("Logged 2 flow files"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:test.txt"));
+ REQUIRE(LogTestController::getInstance().contains("key:filename value:.test.txt"));
+}
+
+TEST_CASE("Test if GetFile honors PollInterval property when triggered multiple times between intervals", "[getFileProperty]") {
+ GetFileTestController test_controller;
+ test_controller.setProperty(minifi::processors::GetFile::PollInterval, "100 ms");
+ test_controller.setProperty(minifi::processors::GetFile::KeepSourceFile, "true");
+
+ test_controller.runSession();
+ auto start_time = utils::timeutils::getTimeMillis();
+ while (LogTestController::getInstance().countOccurrences("Logged 2 flow files") < 2) {
+ test_controller.test_plan_->reset();
+ test_controller.runSession();
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+
+ REQUIRE(utils::timeutils::getTimeMillis() - start_time >= 100);
}
[nifi-minifi-cpp] 05/05: MINIFICPP-1222 Remove engine queue from ExecutePythonProcessor
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 4f0dcd838f5afe46545aee2810d16308a9b5cfc5
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Jan 4 16:23:06 2022 +0100
MINIFICPP-1222 Remove engine queue from ExecutePythonProcessor
Originally the Jira ticket was about reowrking the engine queue to only allow the maximum concurrent tasks number of script engines to run at the same time. The problem with that was that even if the script engines were executed concurrently they use the same python interpreter with the global interpreter lock in place so no concurrency can take place.
The only possibility for concurrency to use sub-interpreters which would be also problematic. First of all pybind11 does not support an interface for sub-interpreters, it is only available through the CPython API, but the documentation also warns about the caveats of the sub-interpreters here: https://pybind11.readthedocs.io/en/stable/advanced/embedding.html
Secondly using the CPython API for sub-interpreters would require us to always know which thread we are running on and create separate sub-interpreters and thread states for each thread (in this case it would possibly not even allow us to run the python scripts from the main thread anymore). Managing the thread states and the sub-interpreters would be cumbersome in our architecture.
Because all of the above problems I removed the python script queue from the ExecutePythonProcessor which will run the python scripts on a single thread.
Closes #1227
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.../script/python/ExecutePythonProcessor.cpp | 75 ++++++++--------------
extensions/script/python/ExecutePythonProcessor.h | 30 +++------
2 files changed, 37 insertions(+), 68 deletions(-)
diff --git a/extensions/script/python/ExecutePythonProcessor.cpp b/extensions/script/python/ExecutePythonProcessor.cpp
index 62e8c23..ef09a4e 100644
--- a/extensions/script/python/ExecutePythonProcessor.cpp
+++ b/extensions/script/python/ExecutePythonProcessor.cpp
@@ -79,8 +79,6 @@ void ExecutePythonProcessor::initialize() {
return;
}
- python_logger_ = core::logging::LoggerFactory<ExecutePythonProcessor>::getAliasedLogger(getName());
-
getProperty(ModuleDirectory.getName(), module_directory_);
appendPathForImportModules();
@@ -93,74 +91,44 @@ void ExecutePythonProcessor::initialize() {
// In case of native python processors we require initialization before onSchedule
// so that we can provide manifest of processor identity on C2
- auto engine = getScriptEngine();
- initalizeThroughScriptEngine(*engine);
- handleEngineNoLongerInUse(std::move(engine));
+ python_script_engine_ = createScriptEngine();
+ initalizeThroughScriptEngine();
}
-void ExecutePythonProcessor::initalizeThroughScriptEngine(python::PythonScriptEngine& engine) {
- engine.eval(script_to_exec_);
+void ExecutePythonProcessor::initalizeThroughScriptEngine() {
+ python_script_engine_->eval(script_to_exec_);
auto shared_this = shared_from_this();
- engine.describe(shared_this);
- engine.onInitialize(shared_this);
+ python_script_engine_->describe(shared_this);
+ python_script_engine_->onInitialize(shared_this);
processor_initialized_ = true;
}
void ExecutePythonProcessor::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) {
- std::shared_ptr<python::PythonScriptEngine> engine = getScriptEngine();
if (!processor_initialized_) {
loadScript();
- initalizeThroughScriptEngine(*engine);
+ python_script_engine_ = createScriptEngine();
+ initalizeThroughScriptEngine();
} else {
- reloadScriptIfUsingScriptFileProperty(*engine);
+ reloadScriptIfUsingScriptFileProperty();
if (script_to_exec_.empty()) {
throw std::runtime_error("Neither Script Body nor Script File is available to execute");
}
}
- engine->eval(script_to_exec_);
- engine->onSchedule(context);
-
- handleEngineNoLongerInUse(std::move(engine));
+ gsl_Expects(python_script_engine_);
+ python_script_engine_->eval(script_to_exec_);
+ python_script_engine_->onSchedule(context);
getProperty(ReloadOnScriptChange.getName(), reload_on_script_change_);
}
void ExecutePythonProcessor::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
- auto engine = getScriptEngine();
- reloadScriptIfUsingScriptFileProperty(*engine);
+ reloadScriptIfUsingScriptFileProperty();
if (script_to_exec_.empty()) {
throw std::runtime_error("Neither Script Body nor Script File is available to execute");
}
- engine->onTrigger(context, session);
- handleEngineNoLongerInUse(std::move(engine));
-}
-
-// TODO(hunyadi): This is potentially not what we want. See https://issues.apache.org/jira/browse/MINIFICPP-1222
-std::shared_ptr<python::PythonScriptEngine> ExecutePythonProcessor::getScriptEngine() {
- std::shared_ptr<python::PythonScriptEngine> engine;
- // Use an existing engine, if one is available
- if (script_engine_q_.try_dequeue(engine)) {
- logger_->log_debug("Using available [%p] script engine instance", engine.get());
- return engine;
- }
- engine = createEngine<python::PythonScriptEngine>();
- logger_->log_info("Created new [%p] script engine instance. Number of instances: approx. %d / %d.", engine.get(), script_engine_q_.size_approx(), getMaxConcurrentTasks());
- if (engine == nullptr) {
- throw std::runtime_error("No script engine available");
- }
- return engine;
-}
-
-void ExecutePythonProcessor::handleEngineNoLongerInUse(std::shared_ptr<python::PythonScriptEngine>&& engine) {
- // Make engine available for use again
- if (script_engine_q_.size_approx() < getMaxConcurrentTasks()) {
- logger_->log_debug("Releasing [%p] script engine", engine.get());
- script_engine_q_.enqueue(engine);
- } else {
- logger_->log_info("Destroying script engine because it is no longer needed");
- }
+ python_script_engine_->onTrigger(context, session);
}
void ExecutePythonProcessor::appendPathForImportModules() {
@@ -203,7 +171,7 @@ void ExecutePythonProcessor::loadScript() {
return;
}
-void ExecutePythonProcessor::reloadScriptIfUsingScriptFileProperty(python::PythonScriptEngine& engine) {
+void ExecutePythonProcessor::reloadScriptIfUsingScriptFileProperty() {
if (script_file_path_.empty() || !reload_on_script_change_) {
return;
}
@@ -212,10 +180,21 @@ void ExecutePythonProcessor::reloadScriptIfUsingScriptFileProperty(python::Pytho
logger_->log_debug("Script file has changed since last time, reloading...");
loadScriptFromFile();
last_script_write_time_ = file_write_time;
- engine.eval(script_to_exec_);
+ python_script_engine_->eval(script_to_exec_);
}
}
+std::unique_ptr<PythonScriptEngine> ExecutePythonProcessor::createScriptEngine() {
+ auto engine = std::make_unique<PythonScriptEngine>();
+
+ python_logger_ = core::logging::LoggerFactory<ExecutePythonProcessor>::getAliasedLogger(getName());
+ engine->bind("log", python_logger_);
+ engine->bind("REL_SUCCESS", Success);
+ engine->bind("REL_FAILURE", Failure);
+
+ return engine;
+}
+
REGISTER_RESOURCE(
ExecutePythonProcessor, "Executes a script given the flow file and a process session. The script is responsible for handling the incoming flow file (transfer to SUCCESS or remove, e.g.) "
"as well as any flow files created by the script. If the handling is incomplete or incorrect, the session will be rolled back.Scripts must define an onTrigger function which accepts NiFi Context"
diff --git a/extensions/script/python/ExecutePythonProcessor.h b/extensions/script/python/ExecutePythonProcessor.h
index 14dc48f..a3904ab 100644
--- a/extensions/script/python/ExecutePythonProcessor.h
+++ b/extensions/script/python/ExecutePythonProcessor.h
@@ -48,8 +48,7 @@ class ExecutePythonProcessor : public core::Processor {
processor_initialized_(false),
python_dynamic_(false),
reload_on_script_change_(true),
- last_script_write_time_(0),
- script_engine_q_() {
+ last_script_write_time_(0) {
}
EXTENSIONAPI static const core::Property ScriptFile;
@@ -93,6 +92,10 @@ class ExecutePythonProcessor : public core::Processor {
return false;
}
+ bool isSingleThreaded() const override {
+ return true;
+ }
+
private:
std::vector<core::Property> python_properties_;
@@ -102,35 +105,22 @@ class ExecutePythonProcessor : public core::Processor {
bool python_dynamic_;
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ExecutePythonProcessor>::getLogger();
- std::shared_ptr<core::logging::Logger> python_logger_;
std::string script_to_exec_;
std::string module_directory_;
bool reload_on_script_change_;
uint64_t last_script_write_time_;
std::string script_file_path_;
+ std::shared_ptr<core::logging::Logger> python_logger_;
+ std::unique_ptr<PythonScriptEngine> python_script_engine_;
- moodycamel::ConcurrentQueue<std::shared_ptr<python::PythonScriptEngine>> script_engine_q_;
-
- void initalizeThroughScriptEngine(python::PythonScriptEngine& engine);
- std::shared_ptr<python::PythonScriptEngine> getScriptEngine();
- void handleEngineNoLongerInUse(std::shared_ptr<python::PythonScriptEngine>&& engine);
void appendPathForImportModules();
void loadScriptFromFile();
void loadScript();
- void reloadScriptIfUsingScriptFileProperty(python::PythonScriptEngine& engine);
+ void reloadScriptIfUsingScriptFileProperty();
+ void initalizeThroughScriptEngine();
-
- template<typename T>
- std::shared_ptr<T> createEngine() const {
- auto engine = std::make_shared<T>();
-
- engine->bind("log", python_logger_);
- engine->bind("REL_SUCCESS", Success);
- engine->bind("REL_FAILURE", Failure);
-
- return engine;
- }
+ std::unique_ptr<PythonScriptEngine> createScriptEngine();
};
} /* namespace processors */
[nifi-minifi-cpp] 01/05: MINIFICPP-1698 Make archive read/write agent-wide available
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 24d9f3a53e66c9ea2f3c24f5445c164cce01a727
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Tue Jan 4 15:48:37 2022 +0100
MINIFICPP-1698 Make archive read/write agent-wide available
Closes #1224
Signed-off-by: Marton Szasz <sz...@apache.org>
---
extensions/libarchive/ArchiveStreamProvider.cpp | 49 ++++
extensions/libarchive/CompressContent.cpp | 77 ++++--
extensions/libarchive/CompressContent.h | 276 +--------------------
extensions/libarchive/ReadArchiveStream.cpp | 88 +++++++
extensions/libarchive/ReadArchiveStream.h | 89 +++++++
extensions/libarchive/WriteArchiveStream.cpp | 131 ++++++++++
extensions/libarchive/WriteArchiveStream.h | 86 +++++++
libminifi/include/core/ProcessSession.h | 8 +
libminifi/include/io/ArchiveStream.h | 54 ++++
libminifi/include/io/StreamPipe.h | 28 ++-
.../test/archive-tests/ArchiveStreamTests.cpp | 55 ++++
11 files changed, 646 insertions(+), 295 deletions(-)
diff --git a/extensions/libarchive/ArchiveStreamProvider.cpp b/extensions/libarchive/ArchiveStreamProvider.cpp
new file mode 100644
index 0000000..776a104
--- /dev/null
+++ b/extensions/libarchive/ArchiveStreamProvider.cpp
@@ -0,0 +1,49 @@
+/**
+ *
+ * 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 <memory>
+
+#include "core/Resource.h"
+#include "WriteArchiveStream.h"
+#include "ReadArchiveStream.h"
+
+namespace org::apache::nifi::minifi::io {
+
+class ArchiveStreamProviderImpl : public ArchiveStreamProvider {
+ public:
+ using ArchiveStreamProvider::ArchiveStreamProvider;
+ std::unique_ptr<WriteArchiveStream> createWriteStream(int compress_level, const std::string& compress_format,
+ std::shared_ptr<OutputStream> sink, std::shared_ptr<core::logging::Logger> logger) override {
+ CompressionFormat format = CompressionFormat::parse(compress_format.c_str(), CompressionFormat{});
+ if (!format) {
+ if (logger) {
+ logger->log_error("Unrecognized compression format '%s'", compress_format);
+ }
+ return nullptr;
+ }
+ return std::make_unique<WriteArchiveStreamImpl>(compress_level, format, std::move(sink));
+ }
+
+ std::unique_ptr<ReadArchiveStream> createReadStream(std::shared_ptr<InputStream> archive_stream) override {
+ return std::make_unique<ReadArchiveStreamImpl>(std::move(archive_stream));
+ }
+};
+
+REGISTER_INTERNAL_RESOURCE_AS(ArchiveStreamProviderImpl, ("ArchiveStreamProvider"));
+
+} // namespace org::apache::nifi::minifi::io
diff --git a/extensions/libarchive/CompressContent.cpp b/extensions/libarchive/CompressContent.cpp
index 12e2016..421bc8d 100644
--- a/extensions/libarchive/CompressContent.cpp
+++ b/extensions/libarchive/CompressContent.cpp
@@ -27,6 +27,7 @@
#include "core/ProcessSession.h"
#include "utils/StringUtils.h"
#include "core/Resource.h"
+#include "io/StreamPipe.h"
namespace org {
namespace apache {
@@ -66,19 +67,19 @@ core::Relationship CompressContent::Failure("failure", "FlowFiles will be transf
const std::string CompressContent::TAR_EXT = ".tar";
-const std::map<std::string, CompressContent::CompressionFormat> CompressContent::compressionFormatMimeTypeMap_{
- {"application/gzip", CompressionFormat::GZIP},
- {"application/bzip2", CompressionFormat::BZIP2},
- {"application/x-bzip2", CompressionFormat::BZIP2},
- {"application/x-lzma", CompressionFormat::LZMA},
- {"application/x-xz", CompressionFormat::XZ_LZMA2}
+const std::map<std::string, io::CompressionFormat> CompressContent::compressionFormatMimeTypeMap_{
+ {"application/gzip", io::CompressionFormat::GZIP},
+ {"application/bzip2", io::CompressionFormat::BZIP2},
+ {"application/x-bzip2", io::CompressionFormat::BZIP2},
+ {"application/x-lzma", io::CompressionFormat::LZMA},
+ {"application/x-xz", io::CompressionFormat::XZ_LZMA2}
};
-const std::map<CompressContent::CompressionFormat, std::string> CompressContent::fileExtension_{
- {CompressionFormat::GZIP, ".gz"},
- {CompressionFormat::LZMA, ".lzma"},
- {CompressionFormat::BZIP2, ".bz2"},
- {CompressionFormat::XZ_LZMA2, ".xz"}
+const std::map<io::CompressionFormat, std::string> CompressContent::fileExtension_{
+ {io::CompressionFormat::GZIP, ".gz"},
+ {io::CompressionFormat::LZMA, ".lzma"},
+ {io::CompressionFormat::BZIP2, ".bz2"},
+ {io::CompressionFormat::XZ_LZMA2, ".xz"}
};
void CompressContent::initialize() {
@@ -129,7 +130,7 @@ void CompressContent::onTrigger(const std::shared_ptr<core::ProcessContext> &con
void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flowFile, const std::shared_ptr<core::ProcessSession>& session) {
session->remove(flowFile);
- CompressionFormat compressFormat;
+ io::CompressionFormat compressFormat;
if (compressFormat_ == ExtendedCompressionFormat::USE_MIME_TYPE) {
std::string attr;
flowFile->getAttribute(core::SpecialFlowAttribute::MIME_TYPE, attr);
@@ -147,22 +148,22 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo
return;
}
} else {
- compressFormat = compressFormat_.cast<CompressionFormat>();
+ compressFormat = compressFormat_.cast<io::CompressionFormat>();
}
std::string mimeType = toMimeType(compressFormat);
// Validate
- if (!encapsulateInTar_ && compressFormat != CompressionFormat::GZIP) {
+ if (!encapsulateInTar_ && compressFormat != io::CompressionFormat::GZIP) {
logger_->log_error("non-TAR encapsulated format only supports GZIP compression");
session->transfer(flowFile, Failure);
return;
}
- if (compressFormat == CompressionFormat::BZIP2 && archive_bzlib_version() == nullptr) {
+ if (compressFormat == io::CompressionFormat::BZIP2 && archive_bzlib_version() == nullptr) {
logger_->log_error("%s compression format is requested, but the agent was compiled without BZip2 support", compressFormat.toString());
session->transfer(flowFile, Failure);
return;
}
- if ((compressFormat == CompressionFormat::LZMA || compressFormat == CompressionFormat::XZ_LZMA2) && archive_liblzma_version() == nullptr) {
+ if ((compressFormat == io::CompressionFormat::LZMA || compressFormat == io::CompressionFormat::XZ_LZMA2) && archive_liblzma_version() == nullptr) {
logger_->log_error("%s compression format is requested, but the agent was compiled without LZMA support ", compressFormat.toString());
session->transfer(flowFile, Failure);
return;
@@ -176,9 +177,37 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo
std::shared_ptr<core::FlowFile> result = session->create(flowFile);
bool success = false;
if (encapsulateInTar_) {
- CompressContent::WriteCallback callback(compressMode_, compressLevel_, compressFormat, flowFile, session);
- session->write(result, &callback);
- success = callback.status_ >= 0;
+ std::function<int64_t(const std::shared_ptr<io::InputStream>&, const std::shared_ptr<io::OutputStream>&)> transformer;
+
+ if (compressMode_ == CompressionMode::Compress) {
+ std::string filename;
+ flowFile->getAttribute(core::SpecialFlowAttribute::FILENAME, filename);
+ transformer = [&, filename] (const std::shared_ptr<io::InputStream>& in, const std::shared_ptr<io::OutputStream>& out) -> int64_t {
+ io::WriteArchiveStreamImpl compressor(compressLevel_, compressFormat, out);
+ if (!compressor.newEntry({filename, in->size()})) {
+ return -1;
+ }
+ return internal::pipe(in.get(), &compressor);
+ };
+ } else {
+ transformer = [&] (const std::shared_ptr<io::InputStream>& in, const std::shared_ptr<io::OutputStream>& out) -> int64_t {
+ io::ReadArchiveStreamImpl decompressor(in);
+ if (!decompressor.nextEntry()) {
+ return -1;
+ }
+ return internal::pipe(&decompressor, out.get());
+ };
+ }
+ session->write(result, FunctionOutputStreamCallback([&] (const auto& out) {
+ return session->read(flowFile, FunctionInputStreamCallback([&] (const auto& in) {
+ return transformer(in, out);
+ }));
+ }));
+ // TODO(adebreceni): previous attempt to handle a malformed archive were in vain
+ // as the session->read threw anyway rolling back the flowfile, we should correctly
+ // forward a malformed archive to failure
+ // https://issues.apache.org/jira/browse/MINIFICPP-1708
+ success = true;
} else {
CompressContent::GzipWriteCallback callback(compressMode_, compressLevel_, flowFile, session);
session->write(result, &callback);
@@ -218,12 +247,12 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo
}
}
-std::string CompressContent::toMimeType(CompressionFormat format) {
+std::string CompressContent::toMimeType(io::CompressionFormat format) {
switch (format.value()) {
- case CompressionFormat::GZIP: return "application/gzip";
- case CompressionFormat::BZIP2: return "application/bzip2";
- case CompressionFormat::LZMA: return "application/x-lzma";
- case CompressionFormat::XZ_LZMA2: return "application/x-xz";
+ case io::CompressionFormat::GZIP: return "application/gzip";
+ case io::CompressionFormat::BZIP2: return "application/bzip2";
+ case io::CompressionFormat::LZMA: return "application/x-lzma";
+ case io::CompressionFormat::XZ_LZMA2: return "application/x-xz";
}
throw Exception(GENERAL_EXCEPTION, "Invalid compression format");
}
diff --git a/extensions/libarchive/CompressContent.h b/extensions/libarchive/CompressContent.h
index bd0c3ba..7e9358b 100644
--- a/extensions/libarchive/CompressContent.h
+++ b/extensions/libarchive/CompressContent.h
@@ -39,6 +39,8 @@
#include "utils/Enum.h"
#include "utils/gsl.h"
#include "utils/Export.h"
+#include "WriteArchiveStream.h"
+#include "ReadArchiveStream.h"
namespace org {
namespace apache {
@@ -81,277 +83,11 @@ class CompressContent : public core::Processor {
(Decompress, "decompress")
)
- SMART_ENUM(CompressionFormat,
- (GZIP, "gzip"),
- (LZMA, "lzma"),
- (XZ_LZMA2, "xz-lzma2"),
- (BZIP2, "bzip2")
- )
-
- SMART_ENUM_EXTEND(ExtendedCompressionFormat, CompressionFormat, (GZIP, LZMA, XZ_LZMA2, BZIP2),
+ SMART_ENUM_EXTEND(ExtendedCompressionFormat, io::CompressionFormat, (GZIP, LZMA, XZ_LZMA2, BZIP2),
(USE_MIME_TYPE, "use mime.type attribute")
)
public:
- // Nest Callback Class for read stream from flow for compress
- class ReadCallbackCompress: public InputStreamCallback {
- public:
- ReadCallbackCompress(std::shared_ptr<core::FlowFile> &flow, struct archive *arch, struct archive_entry *entry) :
- flow_(flow), arch_(arch), entry_(entry), status_(0) {
- }
- ~ReadCallbackCompress() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override {
- uint8_t buffer[4096U];
- int64_t ret = 0;
- uint64_t read_size = 0;
-
- ret = archive_write_header(arch_, entry_);
- if (ret != ARCHIVE_OK) {
- logger_->log_error("Compress Content archive error %s", archive_error_string(arch_));
- status_ = -1;
- return -1;
- }
- while (read_size < flow_->getSize()) {
- const auto readret = stream->read(buffer, sizeof(buffer));
- if (io::isError(readret)) {
- status_ = -1;
- return -1;
- }
- if (readret > 0) {
- ret = archive_write_data(arch_, buffer, readret);
- if (ret < 0) {
- logger_->log_error("Compress Content archive error %s", archive_error_string(arch_));
- status_ = -1;
- return -1;
- }
- read_size += gsl::narrow<uint64_t>(ret);
- } else {
- break;
- }
- }
- return gsl::narrow<int64_t>(read_size);
- }
- std::shared_ptr<core::FlowFile> flow_;
- struct archive *arch_;
- struct archive_entry *entry_;
- int status_;
- std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CompressContent>::getLogger();
- };
- // Nest Callback Class for read stream from flow for decompress
- struct ReadCallbackDecompress : InputStreamCallback {
- explicit ReadCallbackDecompress(std::shared_ptr<core::FlowFile> flow) :
- flow_file(std::move(flow)) {
- }
- ~ReadCallbackDecompress() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override {
- stream->seek(offset);
- const auto readRet = stream->read(buffer, sizeof(buffer));
- stream_read_result = readRet;
- if (!io::isError(readRet)) {
- offset += readRet;
- }
- return gsl::narrow<int64_t>(readRet);
- }
- size_t stream_read_result = 0; // read size or error code, to be checked with io::isError
- uint8_t buffer[8192] = {0};
- size_t offset = 0;
- std::shared_ptr<core::FlowFile> flow_file;
- };
- // Nest Callback Class for write stream
- class WriteCallback: public OutputStreamCallback {
- public:
- WriteCallback(CompressionMode compress_mode, int compress_level, CompressionFormat compress_format,
- const std::shared_ptr<core::FlowFile> &flow, const std::shared_ptr<core::ProcessSession> &session) :
- compress_mode_(compress_mode), compress_level_(compress_level), compress_format_(compress_format),
- flow_(flow), session_(session),
- readDecompressCb_(flow) {
- size_ = 0;
- stream_ = nullptr;
- status_ = 0;
- }
- ~WriteCallback() = default;
-
- CompressionMode compress_mode_;
- int compress_level_;
- CompressionFormat compress_format_;
- std::shared_ptr<core::FlowFile> flow_;
- std::shared_ptr<core::ProcessSession> session_;
- std::shared_ptr<io::BaseStream> stream_;
- int64_t size_;
- std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CompressContent>::getLogger();
- CompressContent::ReadCallbackDecompress readDecompressCb_;
- int status_;
-
- static la_ssize_t archive_write(struct archive* /*arch*/, void *context, const void *buff, size_t size) {
- auto* const callback = static_cast<WriteCallback*>(context);
- const auto ret = callback->stream_->write(reinterpret_cast<const uint8_t*>(buff), size);
- if (!io::isError(ret)) callback->size_ += gsl::narrow<int64_t>(ret);
- return io::isError(ret) ? -1 : gsl::narrow<la_ssize_t>(ret);
- }
-
- static la_ssize_t archive_read(struct archive* archive, void *context, const void **buff) {
- auto *callback = reinterpret_cast<WriteCallback *>(context);
- callback->session_->read(callback->flow_, &callback->readDecompressCb_);
- *buff = callback->readDecompressCb_.buffer;
- if (io::isError(callback->readDecompressCb_.stream_read_result)) {
- archive_set_error(archive, EIO, "Error reading flowfile");
- return -1;
- }
- return gsl::narrow<la_ssize_t>(callback->readDecompressCb_.stream_read_result);
- }
-
- static la_int64_t archive_skip(struct archive* /*a*/, void* /*client_data*/, la_int64_t /*request*/) {
- return 0;
- }
-
- void archive_write_log_error_cleanup(struct archive *arch) {
- logger_->log_error("Compress Content archive write error %s", archive_error_string(arch));
- status_ = -1;
- archive_write_free(arch);
- }
-
- void archive_read_log_error_cleanup(struct archive *arch) {
- logger_->log_error("Compress Content archive read error %s", archive_error_string(arch));
- status_ = -1;
- archive_read_free(arch);
- }
-
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) {
- struct archive *arch;
- int r;
-
- if (compress_mode_ == CompressionMode::Compress) {
- arch = archive_write_new();
- if (!arch) {
- status_ = -1;
- return -1;
- }
- r = archive_write_set_format_ustar(arch);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- if (compress_format_ == CompressionFormat::GZIP) {
- r = archive_write_add_filter_gzip(arch);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- std::string option;
- option = "gzip:compression-level=" + std::to_string(compress_level_);
- r = archive_write_set_options(arch, option.c_str());
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- } else if (compress_format_ == CompressionFormat::BZIP2) {
- r = archive_write_add_filter_bzip2(arch);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- } else if (compress_format_ == CompressionFormat::LZMA) {
- r = archive_write_add_filter_lzma(arch);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- } else if (compress_format_ == CompressionFormat::XZ_LZMA2) {
- r = archive_write_add_filter_xz(arch);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- } else {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- r = archive_write_set_bytes_per_block(arch, 0);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- this->stream_ = stream;
- r = archive_write_open(arch, this, NULL, archive_write, NULL);
- if (r != ARCHIVE_OK) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- struct archive_entry *entry = archive_entry_new();
- if (!entry) {
- archive_write_log_error_cleanup(arch);
- return -1;
- }
- std::string fileName;
- flow_->getAttribute(core::SpecialFlowAttribute::FILENAME, fileName);
- archive_entry_set_pathname(entry, fileName.c_str());
- archive_entry_set_size(entry, flow_->getSize());
- archive_entry_set_mode(entry, S_IFREG | 0755);
- ReadCallbackCompress readCb(flow_, arch, entry);
- session_->read(flow_, &readCb);
- if (readCb.status_ < 0) {
- archive_entry_free(entry);
- archive_write_log_error_cleanup(arch);
- status_ = -1;
- return -1;
- }
- archive_entry_free(entry);
- archive_write_close(arch);
- archive_write_free(arch);
- return size_;
- } else {
- arch = archive_read_new();
- if (!arch) {
- status_ = -1;
- return -1;
- }
- r = archive_read_support_format_all(arch);
- if (r != ARCHIVE_OK) {
- archive_read_log_error_cleanup(arch);
- return -1;
- }
- r = archive_read_support_filter_all(arch);
- if (r != ARCHIVE_OK) {
- archive_read_log_error_cleanup(arch);
- return -1;
- }
- this->stream_ = stream;
- r = archive_read_open2(arch, this, NULL, archive_read, archive_skip, NULL);
- if (r != ARCHIVE_OK) {
- archive_read_log_error_cleanup(arch);
- return -1;
- }
- struct archive_entry *entry;
- if (archive_read_next_header(arch, &entry) != ARCHIVE_OK) {
- archive_read_log_error_cleanup(arch);
- return -1;
- }
- int64_t entry_size = archive_entry_size(entry);
- logger_->log_debug("Decompress Content archive entry size %" PRId64, entry_size);
- size_ = 0;
- while (size_ < entry_size) {
- char buffer[8192];
- const auto read_result = archive_read_data(arch, buffer, sizeof(buffer));
- if (read_result < 0) {
- archive_read_log_error_cleanup(arch);
- return -1;
- }
- if (read_result == 0)
- break;
- size_ += read_result;
- const auto write_result = stream_->write(reinterpret_cast<uint8_t*>(buffer), gsl::narrow<size_t>(read_result));
- if (io::isError(write_result)) {
- archive_read_log_error_cleanup(arch);
- return -1;
- }
- }
- archive_read_close(arch);
- archive_read_free(arch);
- return size_;
- }
- }
- };
-
class GzipWriteCallback : public OutputStreamCallback {
public:
GzipWriteCallback(CompressionMode compress_mode, int compress_level, std::shared_ptr<core::FlowFile> flow, std::shared_ptr<core::ProcessSession> session)
@@ -433,7 +169,7 @@ class CompressContent : public core::Processor {
void initialize() override;
private:
- static std::string toMimeType(CompressionFormat format);
+ static std::string toMimeType(io::CompressionFormat format);
void processFlowFile(const std::shared_ptr<core::FlowFile>& flowFile, const std::shared_ptr<core::ProcessSession>& session);
@@ -448,8 +184,8 @@ class CompressContent : public core::Processor {
bool updateFileName_;
bool encapsulateInTar_;
uint32_t batchSize_{1};
- static const std::map<std::string, CompressionFormat> compressionFormatMimeTypeMap_;
- static const std::map<CompressionFormat, std::string> fileExtension_;
+ static const std::map<std::string, io::CompressionFormat> compressionFormatMimeTypeMap_;
+ static const std::map<io::CompressionFormat, std::string> fileExtension_;
};
} /* namespace processors */
diff --git a/extensions/libarchive/ReadArchiveStream.cpp b/extensions/libarchive/ReadArchiveStream.cpp
new file mode 100644
index 0000000..109e0e7
--- /dev/null
+++ b/extensions/libarchive/ReadArchiveStream.cpp
@@ -0,0 +1,88 @@
+/**
+ *
+ * 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 "ReadArchiveStream.h"
+
+namespace org::apache::nifi::minifi::io {
+
+ReadArchiveStreamImpl::archive_ptr ReadArchiveStreamImpl::createReadArchive() {
+ archive_ptr arch{archive_read_new()};
+ if (!arch) {
+ logger_->log_error("Failed to create read archive");
+ return nullptr;
+ }
+
+ int result;
+
+ result = archive_read_support_format_all(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive read support format all error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ result = archive_read_support_filter_all(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive read support filter all error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ result = archive_read_open2(arch.get(), &reader_, nullptr, archive_read, nullptr, nullptr);
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive read open error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ return arch;
+}
+
+std::optional<EntryInfo> ReadArchiveStreamImpl::nextEntry() {
+ if (!arch_) {
+ return std::nullopt;
+ }
+ entry_size_.reset();
+ struct archive_entry *entry;
+ int result = archive_read_next_header(arch_.get(), &entry);
+ if (result != ARCHIVE_OK) {
+ if (result != ARCHIVE_EOF) {
+ logger_->log_error("Archive read next header error %s", archive_error_string(arch_.get()));
+ }
+ return std::nullopt;
+ }
+ entry_size_ = gsl::narrow<size_t>(archive_entry_size(entry));
+ logger_->log_debug("Archive entry size %zu", entry_size_.value());
+ return EntryInfo{archive_entry_pathname(entry), entry_size_.value()};
+}
+
+size_t ReadArchiveStreamImpl::read(uint8_t* buf, size_t len) {
+ if (!arch_ || !entry_size_) {
+ return STREAM_ERROR;
+ }
+
+ if (len == 0) {
+ return 0;
+ }
+ gsl_Expects(buf);
+
+ const la_ssize_t result = archive_read_data(arch_.get(), buf, len);
+ if (result < 0) {
+ logger_->log_error("Archive read data error %s", archive_error_string(arch_.get()));
+ entry_size_.reset();
+ arch_.reset();
+ return STREAM_ERROR;
+ }
+ return result;
+}
+
+} // namespace org::apache::nifi::minifi::io
diff --git a/extensions/libarchive/ReadArchiveStream.h b/extensions/libarchive/ReadArchiveStream.h
new file mode 100644
index 0000000..7ffaa0c
--- /dev/null
+++ b/extensions/libarchive/ReadArchiveStream.h
@@ -0,0 +1,89 @@
+/**
+ *
+ * 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 <utility>
+
+#include "io/OutputStream.h"
+#include "io/ArchiveStream.h"
+#include "core/logging/LoggerConfiguration.h"
+
+#include "archive_entry.h"
+#include "archive.h"
+
+namespace org::apache::nifi::minifi::io {
+
+class ReadArchiveStreamImpl : public ReadArchiveStream {
+ struct archive_read_deleter {
+ int operator()(struct archive* ptr) const {
+ return archive_read_free(ptr);
+ }
+ };
+ using archive_ptr = std::unique_ptr<struct archive, archive_read_deleter>;
+
+ class BufferedReader {
+ public:
+ explicit BufferedReader(std::shared_ptr<InputStream> input) : input_(std::move(input)) {}
+
+ std::optional<gsl::span<const uint8_t>> readChunk() {
+ size_t result = input_->read(buffer_.data(), buffer_.size());
+ if (io::isError(result)) {
+ return std::nullopt;
+ }
+ return gsl::span<const uint8_t>(buffer_.data(), result);
+ }
+
+ private:
+ std::shared_ptr<InputStream> input_;
+ std::array<uint8_t, 4096> buffer_;
+ };
+
+ archive_ptr createReadArchive();
+
+ public:
+ explicit ReadArchiveStreamImpl(std::shared_ptr<InputStream> input) : reader_(std::move(input)) {
+ arch_ = createReadArchive();
+ }
+
+ std::optional<EntryInfo> nextEntry() override;
+
+ using InputStream::read;
+
+ size_t read(uint8_t* buf, size_t len) override;
+
+ private:
+ static la_ssize_t archive_read(struct archive* archive, void *context, const void **buff) {
+ auto* const input = reinterpret_cast<BufferedReader*>(context);
+ auto opt_buffer = input->readChunk();
+ if (!opt_buffer) {
+ archive_set_error(archive, EIO, "Error reading archive");
+ return -1;
+ }
+ *buff = opt_buffer->data();
+ return gsl::narrow<la_ssize_t>(opt_buffer->size());
+ }
+
+ std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ReadArchiveStream>::getLogger();
+ BufferedReader reader_;
+ archive_ptr arch_;
+ std::optional<size_t> entry_size_;
+};
+
+} // namespace org::apache::nifi::minifi::io
diff --git a/extensions/libarchive/WriteArchiveStream.cpp b/extensions/libarchive/WriteArchiveStream.cpp
new file mode 100644
index 0000000..af375bf
--- /dev/null
+++ b/extensions/libarchive/WriteArchiveStream.cpp
@@ -0,0 +1,131 @@
+/**
+ *
+ * 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 "WriteArchiveStream.h"
+
+#include <utility>
+#include <string>
+
+namespace org::apache::nifi::minifi::io {
+
+WriteArchiveStreamImpl::archive_ptr WriteArchiveStreamImpl::createWriteArchive() {
+ archive_ptr arch{archive_write_new()};
+ if (!arch) {
+ logger_->log_error("Failed to create write archive");
+ return nullptr;
+ }
+
+ int result;
+
+ result = archive_write_set_format_ustar(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write set format ustar error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ if (compress_format_ == CompressionFormat::GZIP) {
+ result = archive_write_add_filter_gzip(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write add filter gzip error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ std::string option = "gzip:compression-level=" + std::to_string(compress_level_);
+ result = archive_write_set_options(arch.get(), option.c_str());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write set options error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ } else if (compress_format_ == CompressionFormat::BZIP2) {
+ result = archive_write_add_filter_bzip2(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write add filter bzip2 error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ } else if (compress_format_ == CompressionFormat::LZMA) {
+ result = archive_write_add_filter_lzma(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write add filter lzma error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ } else if (compress_format_ == CompressionFormat::XZ_LZMA2) {
+ result = archive_write_add_filter_xz(arch.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write add filter xz error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ } else {
+ logger_->log_error("Archive write unsupported compression format");
+ return nullptr;
+ }
+ result = archive_write_set_bytes_per_block(arch.get(), 0);
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write set bytes per block error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ result = archive_write_open(arch.get(), sink_.get(), nullptr, archive_write, nullptr);
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write open error %s", archive_error_string(arch.get()));
+ return nullptr;
+ }
+ return arch;
+}
+
+bool WriteArchiveStreamImpl::newEntry(const EntryInfo& info) {
+ if (!arch_) {
+ return false;
+ }
+ arch_entry_.reset(archive_entry_new());
+ if (!arch_entry_) {
+ logger_->log_error("Failed to create archive entry");
+ return false;
+ }
+ archive_entry_set_pathname(arch_entry_.get(), info.filename.c_str());
+ archive_entry_set_size(arch_entry_.get(), info.size);
+ archive_entry_set_mode(arch_entry_.get(), S_IFREG | 0755);
+
+ int result = archive_write_header(arch_.get(), arch_entry_.get());
+ if (result != ARCHIVE_OK) {
+ logger_->log_error("Archive write header error %s", archive_error_string(arch_.get()));
+ return false;
+ }
+ return true;
+}
+
+size_t WriteArchiveStreamImpl::write(const uint8_t* data, size_t len) {
+ if (!arch_ || !arch_entry_) {
+ return STREAM_ERROR;
+ }
+
+ if (len == 0) {
+ return 0;
+ }
+ gsl_Expects(data);
+
+ int result = archive_write_data(arch_.get(), data, len);
+ if (result < 0) {
+ logger_->log_error("Archive write data error %s", archive_error_string(arch_.get()));
+ arch_entry_.reset();
+ arch_.reset();
+ return STREAM_ERROR;
+ }
+
+ return result;
+}
+
+} // namespace org::apache::nifi::minifi::io
+
+
diff --git a/extensions/libarchive/WriteArchiveStream.h b/extensions/libarchive/WriteArchiveStream.h
new file mode 100644
index 0000000..12e175a
--- /dev/null
+++ b/extensions/libarchive/WriteArchiveStream.h
@@ -0,0 +1,86 @@
+/**
+ *
+ * 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 <utility>
+#include <string>
+
+#include "io/ArchiveStream.h"
+#include "archive_entry.h"
+#include "archive.h"
+#include "utils/Enum.h"
+#include "core/Core.h"
+#include "logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::io {
+
+SMART_ENUM(CompressionFormat,
+ (GZIP, "gzip"),
+ (LZMA, "lzma"),
+ (XZ_LZMA2, "xz-lzma2"),
+ (BZIP2, "bzip2")
+)
+
+class WriteArchiveStreamImpl: public WriteArchiveStream {
+ struct archive_write_deleter {
+ int operator()(struct archive* ptr) const {
+ return archive_write_free(ptr);
+ }
+ };
+ using archive_ptr = std::unique_ptr<struct archive, archive_write_deleter>;
+ struct archive_entry_deleter {
+ void operator()(struct archive_entry* ptr) const {
+ archive_entry_free(ptr);
+ }
+ };
+ using archive_entry_ptr = std::unique_ptr<struct archive_entry, archive_entry_deleter>;
+
+ archive_ptr createWriteArchive();
+
+ public:
+ WriteArchiveStreamImpl(int compress_level, CompressionFormat compress_format, std::shared_ptr<OutputStream> sink)
+ : compress_level_(compress_level),
+ compress_format_(compress_format),
+ sink_(std::move(sink)) {
+ arch_ = createWriteArchive();
+ }
+
+ using OutputStream::write;
+
+ bool newEntry(const EntryInfo& info) override;
+
+ size_t write(const uint8_t* data, size_t len) override;
+
+ private:
+ static la_ssize_t archive_write(struct archive* /*arch*/, void *context, const void *buff, size_t size) {
+ auto* const output = static_cast<OutputStream*>(context);
+ const auto ret = output->write(reinterpret_cast<const uint8_t*>(buff), size);
+ return io::isError(ret) ? -1 : gsl::narrow<la_ssize_t>(ret);
+ }
+
+ int compress_level_;
+ CompressionFormat compress_format_;
+ std::shared_ptr<io::OutputStream> sink_;
+ archive_ptr arch_;
+ archive_entry_ptr arch_entry_;
+ std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<WriteArchiveStreamImpl>::getLogger();
+};
+
+} // namespace org::apache::nifi::minifi::io
diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h
index c80351c..af54e95 100644
--- a/libminifi/include/core/ProcessSession.h
+++ b/libminifi/include/core/ProcessSession.h
@@ -88,8 +88,16 @@ class ProcessSession : public ReferenceContainer {
void remove(const std::shared_ptr<core::FlowFile> &flow);
// Execute the given read callback against the content
int64_t read(const std::shared_ptr<core::FlowFile> &flow, InputStreamCallback *callback);
+
+ int64_t read(const std::shared_ptr<core::FlowFile> &flow, InputStreamCallback&& callback) {
+ return read(flow, &callback);
+ }
// Execute the given write callback against the content
void write(const std::shared_ptr<core::FlowFile> &flow, OutputStreamCallback *callback);
+
+ void write(const std::shared_ptr<core::FlowFile> &flow, OutputStreamCallback&& callback) {
+ return write(flow, &callback);
+ }
// Read and write the flow file at the same time (eg. for processing it line by line)
int64_t readWrite(const std::shared_ptr<core::FlowFile> &flow, InputOutputStreamCallback *callback);
// Replace content with buffer
diff --git a/libminifi/include/io/ArchiveStream.h b/libminifi/include/io/ArchiveStream.h
new file mode 100644
index 0000000..def8838
--- /dev/null
+++ b/libminifi/include/io/ArchiveStream.h
@@ -0,0 +1,54 @@
+/**
+ *
+ * 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 "OutputStream.h"
+#include "InputStream.h"
+#include "core/Core.h"
+#include "core/logging/Logger.h"
+
+namespace org::apache::nifi::minifi::io {
+
+struct EntryInfo {
+ std::string filename;
+ size_t size;
+};
+
+class WriteArchiveStream : public OutputStream {
+ public:
+ virtual bool newEntry(const EntryInfo& info) = 0;
+};
+
+class ReadArchiveStream : public InputStream {
+ public:
+ virtual std::optional<EntryInfo> nextEntry() = 0;
+};
+
+class ArchiveStreamProvider : public core::CoreComponent {
+ public:
+ using CoreComponent::CoreComponent;
+ virtual std::unique_ptr<WriteArchiveStream> createWriteStream(int compress_level, const std::string& compress_format,
+ std::shared_ptr<OutputStream> sink, std::shared_ptr<core::logging::Logger> logger) = 0;
+ virtual std::unique_ptr<ReadArchiveStream> createReadStream(std::shared_ptr<InputStream> archive_stream) = 0;
+};
+
+} // namespace org::apache::nifi::minifi::io
diff --git a/libminifi/include/io/StreamPipe.h b/libminifi/include/io/StreamPipe.h
index 0956f8d..bc30aca 100644
--- a/libminifi/include/io/StreamPipe.h
+++ b/libminifi/include/io/StreamPipe.h
@@ -48,9 +48,31 @@ class InputOutputStreamCallback {
virtual int64_t process(const std::shared_ptr<io::BaseStream>& input, const std::shared_ptr<io::BaseStream>& output) = 0;
};
+class FunctionOutputStreamCallback : public OutputStreamCallback {
+ public:
+ explicit FunctionOutputStreamCallback(std::function<int64_t(const std::shared_ptr<io::OutputStream>&)> fn) : fn_(std::move(fn)) {}
+
+ int64_t process(const std::shared_ptr<io::BaseStream>& stream) override {
+ return fn_(stream);
+ }
+ private:
+ std::function<int64_t(const std::shared_ptr<io::OutputStream>&)> fn_;
+};
+
+class FunctionInputStreamCallback : public InputStreamCallback {
+ public:
+ explicit FunctionInputStreamCallback(std::function<int64_t(const std::shared_ptr<io::InputStream>&)> fn) : fn_(std::move(fn)) {}
+
+ int64_t process(const std::shared_ptr<io::BaseStream>& stream) override {
+ return fn_(stream);
+ }
+ private:
+ std::function<int64_t(const std::shared_ptr<io::InputStream>&)> fn_;
+};
+
namespace internal {
-inline int64_t pipe(const std::shared_ptr<io::InputStream>& src, const std::shared_ptr<io::OutputStream>& dst) {
+inline int64_t pipe(io::InputStream* src, io::OutputStream* dst) {
uint8_t buffer[4096U];
int64_t totalTransferred = 0;
while (true) {
@@ -77,6 +99,10 @@ inline int64_t pipe(const std::shared_ptr<io::InputStream>& src, const std::shar
return totalTransferred;
}
+inline int64_t pipe(const std::shared_ptr<io::InputStream>& src, const std::shared_ptr<io::OutputStream>& dst) {
+ return pipe(src.get(), dst.get());
+}
+
} // namespace internal
class InputStreamPipe : public InputStreamCallback {
diff --git a/libminifi/test/archive-tests/ArchiveStreamTests.cpp b/libminifi/test/archive-tests/ArchiveStreamTests.cpp
new file mode 100644
index 0000000..3c8c715
--- /dev/null
+++ b/libminifi/test/archive-tests/ArchiveStreamTests.cpp
@@ -0,0 +1,55 @@
+/**
+ *
+ * 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 "../TestBase.h"
+#include "WriteArchiveStream.h"
+#include "ReadArchiveStream.h"
+
+using namespace minifi;
+
+TEST_CASE("Create and read archive") {
+ std::map<std::string, std::string> files{
+ {"a.txt", "hello, I'm file A"},
+ {"b.txt", "hello, I'm file B"}
+ };
+
+ auto archive = std::make_shared<io::BufferStream>();
+
+ {
+ io::WriteArchiveStreamImpl compressor(9, io::CompressionFormat::GZIP, archive);
+
+ for (const auto& [filename, content] : files) {
+ REQUIRE(compressor.newEntry({filename, content.length()}));
+ REQUIRE(compressor.write(reinterpret_cast<const uint8_t*>(content.data()), content.length()) == content.length());
+ }
+ }
+
+ {
+ io::ReadArchiveStreamImpl decompressor(archive);
+
+ size_t extracted_entries = 0;
+ while (auto info = decompressor.nextEntry()) {
+ ++extracted_entries;
+ std::string file_content;
+ file_content.resize(info->size);
+ REQUIRE(decompressor.read(reinterpret_cast<uint8_t*>(file_content.data()), file_content.length()) == file_content.length());
+ REQUIRE(files[info->filename] == file_content);
+ }
+ REQUIRE(extracted_entries == files.size());
+ }
+}
[nifi-minifi-cpp] 02/05: MINIFICPP-1290 Create test coverage for OPC processors
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit c4769dd2e862f4716b00359a1a78cbd743afc7b9
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Jan 4 15:57:58 2022 +0100
MINIFICPP-1290 Create test coverage for OPC processors
- Upgraded open6241 library to version 1.2.2 and included the required fixes
- Added integration tests for put and fetch OPC processors with no security, secure connection and authentication.
Note: The docker image used for testing is currently a private lordgamez/open62541 image until the changes included in that image are merged and the new image is uploaded in the official open62541 docker repository. The PR for the changes can be checked here: https://github.com/open62541/open62541/pull/4722
Closes #1216
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.github/workflows/ci.yml | 2 +-
PROCESSORS.md | 4 +-
cmake/BundledOpen62541.cmake | 4 +-
.../integration/MiNiFi_integration_test_driver.py | 21 +--
docker/test/integration/features/opcua.feature | 155 +++++++++++++++++++++
.../minifi/core/AzureStorageServerContainer.py | 7 +-
docker/test/integration/minifi/core/Container.py | 3 +-
.../minifi/core/DockerTestDirectoryBindings.py | 1 +
.../integration/minifi/core/FileSystemObserver.py | 37 ++---
.../test/integration/minifi/core/FlowContainer.py | 4 +-
.../integration/minifi/core/HttpProxyContainer.py | 7 +-
.../minifi/core/KafkaBrokerContainer.py | 7 +-
.../integration/minifi/core/MinifiContainer.py | 8 +-
.../integration/minifi/core/MqttBrokerContainer.py | 7 +-
.../test/integration/minifi/core/NifiContainer.py | 13 +-
.../minifi/core/OPCUAServerContainer.py | 24 ++++
.../minifi/core/PostgreSQLServerContainer.py | 7 +-
.../integration/minifi/core/S3ServerContainer.py | 7 +-
.../minifi/core/SingleNodeDockerCluster.py | 23 +--
.../integration/minifi/core/ZookeeperContainer.py | 7 +-
.../minifi/processors/FetchOPCProcessor.py | 8 ++
.../minifi/processors/PutOPCProcessor.py | 9 ++
.../resources/opcua/opcua_client_cert.der | Bin 0 -> 1018 bytes
.../resources/opcua/opcua_client_key.der | Bin 0 -> 1192 bytes
docker/test/integration/steps/steps.py | 38 ++++-
extensions/opc/include/fetchopc.h | 4 +
extensions/opc/include/opc.h | 4 +-
extensions/opc/include/putopc.h | 4 +
extensions/opc/src/opc.cpp | 68 ++++-----
extensions/opc/src/opcbase.cpp | 61 ++++----
extensions/opc/src/putopc.cpp | 16 +--
thirdparty/open62541/open62541.patch | 50 +++----
32 files changed, 419 insertions(+), 191 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index a1dff22..a933950 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -201,7 +201,7 @@ jobs:
if [ -d ~/.ccache ]; then mv ~/.ccache .; fi
mkdir build
cd build
- cmake -DUSE_SHARED_LIBS= -DSTRICT_GSL_CHECKS=AUDIT -DENABLE_JNI=OFF -DDISABLE_JEMALLOC=ON -DENABLE_AWS=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_AZURE=ON -DENABLE_SQL=ON -DDOCKER_BUILD_ONLY=ON -DDOCKER_CCACHE_DUMP_LOCATION=$HOME/.ccache ..
+ cmake -DUSE_SHARED_LIBS= -DSTRICT_GSL_CHECKS=AUDIT -DENABLE_JNI=OFF -DDISABLE_JEMALLOC=ON -DENABLE_AWS=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_AZURE=ON -DENABLE_SQL=ON -DENABLE_OPC=ON -DDOCKER_BUILD_ONLY=ON -DDOCKER_CCACHE_DUMP_LOCATION=$HOME/.ccache ..
make docker
- id: install_deps
run: |
diff --git a/PROCESSORS.md b/PROCESSORS.md
index fb796af..2445d48 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -523,7 +523,7 @@ In the list below, the names of required properties appear in bold. Any other pr
|**Node ID**|||Specifies the ID of the root node to traverse|
|**Node ID type**||Int<br>Path<br>String<br>|Specifies the type of the provided node ID|
|**OPC server endpoint**|||Specifies the address, port and relative path of an OPC endpoint|
-|Password|||Password to log in with. Providing this requires cert and key to be provided as well, credentials are always sent encrypted.|
+|Password|||Password to log in with.|
|Trusted server certificate path|||Path to the DER-encoded trusted server certificate|
|Username|||Username to log in with.|
### Relationships
@@ -1369,7 +1369,7 @@ In the list below, the names of required properties appear in bold. Any other pr
|**Parent node ID**|||Specifies the ID of the root node to traverse|
|**Parent node ID type**||Int<br>Path<br>String<br>|Specifies the type of the provided node ID|
|Parent node namespace index|0||The index of the namespace. Used only if node ID type is not path.|
-|Password|||Password to log in with. Providing this requires cert and key to be provided as well, credentials are always sent encrypted.|
+|Password|||Password to log in with.|
|Target node ID|||ID of target node.<br/>**Supports Expression Language: true**|
|Target node ID type|||ID type of target node. Allowed values are: Int, String.<br/>**Supports Expression Language: true**|
|Target node browse name|||Browse name of target node. Only used when new node is created.<br/>**Supports Expression Language: true**|
diff --git a/cmake/BundledOpen62541.cmake b/cmake/BundledOpen62541.cmake
index e055d39..c03c0bc 100644
--- a/cmake/BundledOpen62541.cmake
+++ b/cmake/BundledOpen62541.cmake
@@ -40,8 +40,8 @@ function(use_bundled_open62541 SOURCE_DIR BINARY_DIR)
# Build project
ExternalProject_Add(
open62541-external
- URL "https://github.com/open62541/open62541/archive/v1.0.tar.gz"
- URL_HASH "SHA256=9be66efefe2cdb07a7638aad91c301b5c6163f99c66995bc41cce31ec0ea207e"
+ URL "https://github.com/open62541/open62541/archive/v1.2.2.tar.gz"
+ URL_HASH "SHA256=9b5bfd811ee523be601f11abc514a93c67fe5c6e957cd6c36fe6ea4f28e009bb"
SOURCE_DIR "${BINARY_DIR}/thirdparty/open62541-src"
PATCH_COMMAND ${PC}
LIST_SEPARATOR % # This is needed for passing semicolon-separated lists
diff --git a/docker/test/integration/MiNiFi_integration_test_driver.py b/docker/test/integration/MiNiFi_integration_test_driver.py
index b7e5991..34bc202 100644
--- a/docker/test/integration/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/MiNiFi_integration_test_driver.py
@@ -44,8 +44,8 @@ class MiNiFi_integration_test():
def docker_path_to_local_path(self, docker_path):
return self.docker_directory_bindings.docker_path_to_local_path(self.test_id, docker_path)
- def acquire_container(self, name, engine='minifi-cpp'):
- return self.cluster.acquire_container(name, engine)
+ def acquire_container(self, name, engine='minifi-cpp', command=None):
+ return self.cluster.acquire_container(name, engine, command)
def wait_for_container_startup_to_finish(self, container_name):
startup_success = self.cluster.wait_for_startup_log(container_name, 120)
@@ -141,7 +141,8 @@ class MiNiFi_integration_test():
def check_for_at_least_one_file_with_content_generated(self, content, timeout_seconds):
output_validator = SingleOrMultiFileOutputValidator(decode_escaped_str(content))
output_validator.set_output_dir(self.file_system_observer.get_output_dir())
- self.check_output(timeout_seconds, output_validator, 1)
+ expected_number_of_files = timeout_seconds
+ self.check_output(timeout_seconds, output_validator, expected_number_of_files)
def check_for_num_files_generated(self, num_flowfiles, timeout_seconds):
output_validator = NoContentCheckFileNumberValidator(num_flowfiles)
@@ -163,7 +164,8 @@ class MiNiFi_integration_test():
self.validate(output_validator)
def check_output(self, timeout_seconds, output_validator, max_files):
- self.file_system_observer.wait_for_output(timeout_seconds, max_files)
+ if self.file_system_observer.wait_for_output(timeout_seconds, max_files, output_validator):
+ return
self.validate(output_validator)
def validate(self, validator):
@@ -190,17 +192,20 @@ class MiNiFi_integration_test():
assert self.cluster.wait_for_kafka_consumer_to_be_registered(kafka_container_name)
def check_minifi_log_contents(self, line, timeout_seconds=60):
+ self.check_container_log_contents("minifi-cpp", line, timeout_seconds)
+
+ def check_minifi_log_matches_regex(self, regex, timeout_seconds=60):
for container in self.cluster.containers.values():
if container.get_engine() == "minifi-cpp":
- line_found = self.cluster.wait_for_app_logs(container.get_name(), line, timeout_seconds)
+ line_found = self.cluster.wait_for_app_logs_regex(container.get_name(), regex, timeout_seconds)
if line_found:
return
assert False
- def check_minifi_log_matches_regex(self, regex, timeout_seconds=60):
+ def check_container_log_contents(self, container_engine, line, timeout_seconds=60):
for container in self.cluster.containers.values():
- if container.get_engine() == "minifi-cpp":
- line_found = self.cluster.wait_for_app_logs_regex(container.get_name(), regex, timeout_seconds)
+ if container.get_engine() == container_engine:
+ line_found = self.cluster.wait_for_app_logs(container.get_name(), line, timeout_seconds)
if line_found:
return
assert False
diff --git a/docker/test/integration/features/opcua.feature b/docker/test/integration/features/opcua.feature
new file mode 100644
index 0000000..8e44aee
--- /dev/null
+++ b/docker/test/integration/features/opcua.feature
@@ -0,0 +1,155 @@
+Feature: Putting and fetching data to OPC UA server
+ In order to send and fetch data from an OPC UA server
+ As a user of MiNiFi
+ I need to have PutOPCProcessor and FetchOPCProcessor
+
+ Background:
+ Given the content of "/tmp/output" is monitored
+
+ Scenario Outline: Create and fetch data from an OPC UA node
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "create-opc-ua-node" flow
+ And a file with the content "<Value>" is present in "/tmp/input"
+ And a PutOPCProcessor processor in the "create-opc-ua-node" flow
+ And a FetchOPCProcessor processor in the "fetch-opc-ua-node" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "fetch-opc-ua-node" flow
+ And these processor properties are set:
+ | processor name | property name | property value |
+ | PutOPCProcessor | Parent node ID | 85 |
+ | PutOPCProcessor | Parent node ID type | Int |
+ | PutOPCProcessor | Target node ID | 9999 |
+ | PutOPCProcessor | Target node ID type | Int |
+ | PutOPCProcessor | Target node namespace index | 1 |
+ | PutOPCProcessor | Value type | <Value Type> |
+ | PutOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | PutOPCProcessor | Target node browse name | testnodename |
+ | FetchOPCProcessor | Node ID | 9999 |
+ | FetchOPCProcessor | Node ID type | Int |
+ | FetchOPCProcessor | Namespace index | 1 |
+ | FetchOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | FetchOPCProcessor | Max depth | 1 |
+
+ And the "success" relationship of the GetFile processor is connected to the PutOPCProcessor
+ And the "success" relationship of the FetchOPCProcessor processor is connected to the PutFile
+
+ And an OPC UA server is set up
+
+ When all instances start up
+ Then at least one flowfile with the content "<Value>" is placed in the monitored directory in less than 60 seconds
+
+ Examples: Topic names and formats to test
+ | Value Type | Value |
+ | String | Test |
+ | UInt32 | 42 |
+ | Double | 123.321 |
+ | Boolean | False |
+
+ Scenario Outline: Update and fetch data from an OPC UA node
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "update-opc-ua-node" flow
+ And a file with the content "<Value>" is present in "/tmp/input"
+ And a PutOPCProcessor processor in the "update-opc-ua-node" flow
+ And a FetchOPCProcessor processor in the "fetch-opc-ua-node" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "fetch-opc-ua-node" flow
+ And these processor properties are set:
+ | processor name | property name | property value |
+ | PutOPCProcessor | Parent node ID | 85 |
+ | PutOPCProcessor | Parent node ID type | Int |
+ | PutOPCProcessor | Target node ID | <Node ID> |
+ | PutOPCProcessor | Target node ID type | <Node ID Type> |
+ | PutOPCProcessor | Target node namespace index | 1 |
+ | PutOPCProcessor | Value type | <Value Type> |
+ | PutOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | PutOPCProcessor | Target node browse name | testnodename |
+ | FetchOPCProcessor | Node ID | <Node ID> |
+ | FetchOPCProcessor | Node ID type | <Node ID Type> |
+ | FetchOPCProcessor | Namespace index | 1 |
+ | FetchOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | FetchOPCProcessor | Max depth | 1 |
+
+ And the "success" relationship of the GetFile processor is connected to the PutOPCProcessor
+ And the "success" relationship of the FetchOPCProcessor processor is connected to the PutFile
+
+ And an OPC UA server is set up
+
+ When all instances start up
+ Then at least one flowfile with the content "<Value>" is placed in the monitored directory in less than 60 seconds
+
+ # Node ids starting from 51000 are pre-defined demo node ids in the test server application (server_ctt) of the open62541 docker image. There is one nodeid defined
+ # for each type supported by OPC UA. These demo nodes can be used for testing purposes. "the.answer" is also a pre-defined string id for the same testing purposes.
+ Examples: Topic names and formats to test
+ | Node ID Type | Node ID | Value Type | Value |
+ | Int | 51034 | String | minifi-test |
+ | Int | 51001 | Boolean | True |
+ | String | the.answer | Int32 | 54 |
+ | Int | 51019 | UInt32 | 123 |
+ | Int | 51031 | Double | 66.6 |
+
+ Scenario: Create and fetch data from an OPC UA node through secure connection
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "create-opc-ua-node" flow
+ And a file with the content "Test" is present in "/tmp/input"
+ And a PutOPCProcessor processor in the "create-opc-ua-node" flow
+ And a FetchOPCProcessor processor in the "fetch-opc-ua-node" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "fetch-opc-ua-node" flow
+ And these processor properties are set:
+ | processor name | property name | property value |
+ | PutOPCProcessor | Parent node ID | 85 |
+ | PutOPCProcessor | Parent node ID type | Int |
+ | PutOPCProcessor | Target node ID | 9999 |
+ | PutOPCProcessor | Target node ID type | Int |
+ | PutOPCProcessor | Target node namespace index | 1 |
+ | PutOPCProcessor | Value type | String |
+ | PutOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | PutOPCProcessor | Target node browse name | testnodename |
+ | PutOPCProcessor | Certificate path | /tmp/resources/opcua/opcua_client_cert.der |
+ | PutOPCProcessor | Key path | /tmp/resources/opcua/opcua_client_key.der |
+ | PutOPCProcessor | Application URI | urn:open62541.server.application |
+ | FetchOPCProcessor | Node ID | 9999 |
+ | FetchOPCProcessor | Node ID type | Int |
+ | FetchOPCProcessor | Namespace index | 1 |
+ | FetchOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | FetchOPCProcessor | Max depth | 1 |
+ | FetchOPCProcessor | Certificate path | /tmp/resources/opcua/opcua_client_cert.der |
+ | FetchOPCProcessor | Key path | /tmp/resources/opcua/opcua_client_key.der |
+ | FetchOPCProcessor | Application URI | urn:open62541.server.application |
+
+ And the "success" relationship of the GetFile processor is connected to the PutOPCProcessor
+ And the "success" relationship of the FetchOPCProcessor processor is connected to the PutFile
+
+ And an OPC UA server is set up
+
+ When all instances start up
+ Then at least one flowfile with the content "Test" is placed in the monitored directory in less than 60 seconds
+ And the OPC UA server logs contain the following message: "SecureChannel opened with SecurityPolicy http://opcfoundation.org/UA/SecurityPolicy#Aes128_Sha256_RsaOaep" in less than 5 seconds
+
+ Scenario: Create and fetch data from an OPC UA node through username and password authenticated connection
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input" in the "create-opc-ua-node" flow
+ And a file with the content "Test" is present in "/tmp/input"
+ And a PutOPCProcessor processor in the "create-opc-ua-node" flow
+ And a FetchOPCProcessor processor in the "fetch-opc-ua-node" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "fetch-opc-ua-node" flow
+ And these processor properties are set:
+ | processor name | property name | property value |
+ | PutOPCProcessor | Parent node ID | 85 |
+ | PutOPCProcessor | Parent node ID type | Int |
+ | PutOPCProcessor | Target node ID | 9999 |
+ | PutOPCProcessor | Target node ID type | Int |
+ | PutOPCProcessor | Target node namespace index | 1 |
+ | PutOPCProcessor | Value type | String |
+ | PutOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | PutOPCProcessor | Target node browse name | testnodename |
+ | PutOPCProcessor | Username | peter |
+ | PutOPCProcessor | Password | peter123 |
+ | FetchOPCProcessor | Node ID | 9999 |
+ | FetchOPCProcessor | Node ID type | Int |
+ | FetchOPCProcessor | Namespace index | 1 |
+ | FetchOPCProcessor | OPC server endpoint | opc.tcp://opcua-server:4840/ |
+ | FetchOPCProcessor | Max depth | 1 |
+ | FetchOPCProcessor | Username | peter |
+ | FetchOPCProcessor | Password | peter123 |
+
+ And the "success" relationship of the GetFile processor is connected to the PutOPCProcessor
+ And the "success" relationship of the FetchOPCProcessor processor is connected to the PutFile
+
+ And an OPC UA server is set up with access control
+
+ When all instances start up
+ Then at least one flowfile with the content "Test" is placed in the monitored directory in less than 60 seconds
diff --git a/docker/test/integration/minifi/core/AzureStorageServerContainer.py b/docker/test/integration/minifi/core/AzureStorageServerContainer.py
index 6e8645d..798a04d 100644
--- a/docker/test/integration/minifi/core/AzureStorageServerContainer.py
+++ b/docker/test/integration/minifi/core/AzureStorageServerContainer.py
@@ -3,8 +3,8 @@ from .Container import Container
class AzureStorageServerContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 'azure-storage-server', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'azure-storage-server', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "Azurite Queue service is successfully listening at"
@@ -19,5 +19,6 @@ class AzureStorageServerContainer(Container):
detach=True,
name=self.name,
network=self.network.name,
- ports={'10000/tcp': 10000, '10001/tcp': 10001})
+ ports={'10000/tcp': 10000, '10001/tcp': 10001},
+ entrypoint=self.command)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/Container.py b/docker/test/integration/minifi/core/Container.py
index 8e0b41c..1457230 100644
--- a/docker/test/integration/minifi/core/Container.py
+++ b/docker/test/integration/minifi/core/Container.py
@@ -3,12 +3,13 @@ import logging
class Container:
- def __init__(self, name, engine, vols, network, image_store):
+ def __init__(self, name, engine, vols, network, image_store, command):
self.name = name
self.engine = engine
self.vols = vols
self.network = network
self.image_store = image_store
+ self.command = command
# Get docker client
self.client = docker.from_env()
diff --git a/docker/test/integration/minifi/core/DockerTestDirectoryBindings.py b/docker/test/integration/minifi/core/DockerTestDirectoryBindings.py
index 1aed63a..1673327 100644
--- a/docker/test/integration/minifi/core/DockerTestDirectoryBindings.py
+++ b/docker/test/integration/minifi/core/DockerTestDirectoryBindings.py
@@ -25,6 +25,7 @@ class DockerTestDirectoryBindings:
test_dir = os.environ['TEST_DIRECTORY'] # Based on DockerVerify.sh
shutil.copytree(test_dir + "/resources/kafka_broker/conf/certs", self.data_directories[test_id]["resources_dir"] + "/certs")
shutil.copytree(test_dir + "/resources/python", self.data_directories[test_id]["resources_dir"] + "/python")
+ shutil.copytree(test_dir + "/resources/opcua", self.data_directories[test_id]["resources_dir"] + "/opcua")
def get_data_directories(self, test_id):
return self.data_directories[test_id]
diff --git a/docker/test/integration/minifi/core/FileSystemObserver.py b/docker/test/integration/minifi/core/FileSystemObserver.py
index 0411bf5..329acc6 100644
--- a/docker/test/integration/minifi/core/FileSystemObserver.py
+++ b/docker/test/integration/minifi/core/FileSystemObserver.py
@@ -31,21 +31,26 @@ class FileSystemObserver(object):
self.observer.schedule(self.event_handler, self.test_output_dir, recursive=True)
self.observer.start()
- def wait_for_output(self, timeout_seconds, max_files):
+ def wait_for_output(self, timeout_seconds, max_files, output_validator):
logging.info('Waiting up to %d seconds for %d test outputs...', timeout_seconds, max_files)
self.restart_observer_if_needed()
- if max_files <= self.event_handler.get_num_files_created():
- return
- wait_start_time = time.perf_counter()
- for _ in range(0, max_files):
- # Note: The timing on Event.wait() is inaccurate
- self.done_event.wait(timeout_seconds - time.perf_counter() + wait_start_time)
- if self.done_event.isSet():
- self.done_event.clear()
- if max_files <= self.event_handler.get_num_files_created():
- self.done_event.set()
- return
- if timeout_seconds < (time.perf_counter() - wait_start_time):
- break
- self.observer.stop()
- self.observer.join()
+ try:
+ if max_files <= self.event_handler.get_num_files_created():
+ return False
+ wait_start_time = time.perf_counter()
+ for _ in range(0, max_files):
+ # Note: The timing on Event.wait() is inaccurate
+ self.done_event.wait(timeout_seconds - time.perf_counter() + wait_start_time)
+ if self.done_event.isSet():
+ self.done_event.clear()
+ if max_files <= self.event_handler.get_num_files_created():
+ self.done_event.set()
+ return False
+ if output_validator.validate():
+ return True
+ if timeout_seconds < (time.perf_counter() - wait_start_time):
+ break
+ return False
+ finally:
+ self.observer.stop()
+ self.observer.join()
diff --git a/docker/test/integration/minifi/core/FlowContainer.py b/docker/test/integration/minifi/core/FlowContainer.py
index f8dfc23..d8eaffa 100644
--- a/docker/test/integration/minifi/core/FlowContainer.py
+++ b/docker/test/integration/minifi/core/FlowContainer.py
@@ -2,8 +2,8 @@ from .Container import Container
class FlowContainer(Container):
- def __init__(self, config_dir, name, engine, vols, network, image_store):
- super().__init__(name, engine, vols, network, image_store)
+ def __init__(self, config_dir, name, engine, vols, network, image_store, command):
+ super().__init__(name, engine, vols, network, image_store, command)
self.start_nodes = []
self.config_dir = config_dir
diff --git a/docker/test/integration/minifi/core/HttpProxyContainer.py b/docker/test/integration/minifi/core/HttpProxyContainer.py
index 98fe2a9..79e5872 100644
--- a/docker/test/integration/minifi/core/HttpProxyContainer.py
+++ b/docker/test/integration/minifi/core/HttpProxyContainer.py
@@ -3,8 +3,8 @@ from .Container import Container
class HttpProxyContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 'http-proxy', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'http-proxy', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "Accepting HTTP Socket connections at"
@@ -19,5 +19,6 @@ class HttpProxyContainer(Container):
detach=True,
name=self.name,
network=self.network.name,
- ports={'3128/tcp': 3128})
+ ports={'3128/tcp': 3128},
+ entrypoint=self.command)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/KafkaBrokerContainer.py b/docker/test/integration/minifi/core/KafkaBrokerContainer.py
index 64346c3..fcc6647 100644
--- a/docker/test/integration/minifi/core/KafkaBrokerContainer.py
+++ b/docker/test/integration/minifi/core/KafkaBrokerContainer.py
@@ -3,8 +3,8 @@ from .Container import Container
class KafkaBrokerContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 'kafka-broker', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'kafka-broker', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "Kafka startTimeMs"
@@ -30,5 +30,6 @@ class KafkaBrokerContainer(Container):
"KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka-broker:9092,PLAINTEXT_HOST://localhost:29092,SSL://kafka-broker:9093,SSL_HOST://localhost:29093,SASL_PLAINTEXT://kafka-broker:9094,SASL_PLAINTEXT_HOST://localhost:29094,SASL_SSL://kafka-broker:9095,SASL_SSL_HOST://localhost:29095",
"KAFKA_HEAP_OPTS=-Xms512m -Xmx1g",
"KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181",
- "SSL_CLIENT_AUTH=none"])
+ "SSL_CLIENT_AUTH=none"],
+ entrypoint=self.command)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/MinifiContainer.py b/docker/test/integration/minifi/core/MinifiContainer.py
index c1b3758..8b971bf 100644
--- a/docker/test/integration/minifi/core/MinifiContainer.py
+++ b/docker/test/integration/minifi/core/MinifiContainer.py
@@ -8,8 +8,10 @@ class MinifiContainer(FlowContainer):
MINIFI_VERSION = os.environ['MINIFI_VERSION']
MINIFI_ROOT = '/opt/minifi/nifi-minifi-cpp-' + MINIFI_VERSION
- def __init__(self, config_dir, name, vols, network, image_store):
- super().__init__(config_dir, name, 'minifi-cpp', vols, network, image_store)
+ def __init__(self, config_dir, name, vols, network, image_store, command=None):
+ if not command:
+ command = ["/bin/sh", "-c", "cp /tmp/minifi_config/config.yml " + MinifiContainer.MINIFI_ROOT + "/conf && /opt/minifi/minifi-current/bin/minifi.sh run"]
+ super().__init__(config_dir, name, 'minifi-cpp', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "Starting Flow Controller"
@@ -36,6 +38,6 @@ class MinifiContainer(FlowContainer):
detach=True,
name=self.name,
network=self.network.name,
- entrypoint=["/bin/sh", "-c", "cp /tmp/minifi_config/config.yml " + MinifiContainer.MINIFI_ROOT + "/conf && /opt/minifi/minifi-current/bin/minifi.sh run"],
+ entrypoint=self.command,
volumes=self.vols)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/MqttBrokerContainer.py b/docker/test/integration/minifi/core/MqttBrokerContainer.py
index 51632b4..7900e27 100644
--- a/docker/test/integration/minifi/core/MqttBrokerContainer.py
+++ b/docker/test/integration/minifi/core/MqttBrokerContainer.py
@@ -3,8 +3,8 @@ from .Container import Container
class MqttBrokerContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 'mqtt-broker', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'mqtt-broker', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "mosquitto version [0-9\\.]+ running"
@@ -19,5 +19,6 @@ class MqttBrokerContainer(Container):
detach=True,
name=self.name,
network=self.network.name,
- ports={'1883/tcp': 1883})
+ ports={'1883/tcp': 1883},
+ entrypoint=self.command)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/NifiContainer.py b/docker/test/integration/minifi/core/NifiContainer.py
index 6475133..42bdf48 100644
--- a/docker/test/integration/minifi/core/NifiContainer.py
+++ b/docker/test/integration/minifi/core/NifiContainer.py
@@ -10,8 +10,12 @@ class NifiContainer(FlowContainer):
NIFI_VERSION = '1.7.0'
NIFI_ROOT = '/opt/nifi/nifi-' + NIFI_VERSION
- def __init__(self, config_dir, name, vols, network, image_store):
- super().__init__(config_dir, name, 'nifi', vols, network, image_store)
+ def __init__(self, config_dir, name, vols, network, image_store, command=None):
+ if not command:
+ entry_command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
+ r"cp /tmp/nifi_config/flow.xml.gz {nifi_root}/conf && /opt/nifi/scripts/start.sh").format(name=name, nifi_root=NifiContainer.NIFI_ROOT)
+ command = ["/bin/sh", "-c", entry_command]
+ super().__init__(config_dir, name, 'nifi', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "Starting Flow Controller"
@@ -33,15 +37,12 @@ class NifiContainer(FlowContainer):
logging.info('Creating and running nifi docker container...')
self.__create_config()
-
- command = (r"sed -i -e 's/^\(nifi.remote.input.host\)=.*/\1={name}/' {nifi_root}/conf/nifi.properties && "
- r"cp /tmp/nifi_config/flow.xml.gz {nifi_root}/conf && /opt/nifi/scripts/start.sh").format(name=self.name, nifi_root=NifiContainer.NIFI_ROOT)
self.client.containers.run(
self.image_store.get_image(self.get_engine()),
detach=True,
name=self.name,
hostname=self.name,
network=self.network.name,
- entrypoint=["/bin/sh", "-c", command],
+ entrypoint=self.command,
volumes=self.vols)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/OPCUAServerContainer.py b/docker/test/integration/minifi/core/OPCUAServerContainer.py
new file mode 100644
index 0000000..969395c
--- /dev/null
+++ b/docker/test/integration/minifi/core/OPCUAServerContainer.py
@@ -0,0 +1,24 @@
+import logging
+from .Container import Container
+
+
+class OPCUAServerContainer(Container):
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'opcua-server', vols, network, image_store, command)
+
+ def get_startup_finished_log_entry(self):
+ return "TCP network layer listening on"
+
+ def deploy(self):
+ if not self.set_deployed():
+ return
+
+ logging.info('Creating and running OPC UA server docker container...')
+ self.client.containers.run(
+ "lordgamez/open62541",
+ detach=True,
+ name=self.name,
+ network=self.network.name,
+ ports={'4840/tcp': 4840},
+ entrypoint=self.command)
+ logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/PostgreSQLServerContainer.py b/docker/test/integration/minifi/core/PostgreSQLServerContainer.py
index 97c7e6c..90c3700 100644
--- a/docker/test/integration/minifi/core/PostgreSQLServerContainer.py
+++ b/docker/test/integration/minifi/core/PostgreSQLServerContainer.py
@@ -2,8 +2,8 @@ from .Container import Container
class PostgreSQLServerContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 'postgresql-server', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'postgresql-server', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "database system is ready to accept connections"
@@ -17,4 +17,5 @@ class PostgreSQLServerContainer(Container):
detach=True,
name='postgresql-server',
network=self.network.name,
- environment=["POSTGRES_PASSWORD=password"])
+ environment=["POSTGRES_PASSWORD=password"],
+ entrypoint=self.command)
diff --git a/docker/test/integration/minifi/core/S3ServerContainer.py b/docker/test/integration/minifi/core/S3ServerContainer.py
index 0ddd9a9..565be4d 100644
--- a/docker/test/integration/minifi/core/S3ServerContainer.py
+++ b/docker/test/integration/minifi/core/S3ServerContainer.py
@@ -3,8 +3,8 @@ from .Container import Container
class S3ServerContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 's3-server', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command):
+ super().__init__(name, 's3-server', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "Started S3MockApplication"
@@ -20,5 +20,6 @@ class S3ServerContainer(Container):
name=self.name,
network=self.network.name,
ports={'9090/tcp': 9090, '9191/tcp': 9191},
- environment=["initialBuckets=test_bucket"])
+ environment=["initialBuckets=test_bucket"],
+ entrypoint=self.command)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/core/SingleNodeDockerCluster.py b/docker/test/integration/minifi/core/SingleNodeDockerCluster.py
index 631c7b3..59e6ba4 100644
--- a/docker/test/integration/minifi/core/SingleNodeDockerCluster.py
+++ b/docker/test/integration/minifi/core/SingleNodeDockerCluster.py
@@ -12,6 +12,7 @@ from .AzureStorageServerContainer import AzureStorageServerContainer
from .HttpProxyContainer import HttpProxyContainer
from .PostgreSQLServerContainer import PostgreSQLServerContainer
from .MqttBrokerContainer import MqttBrokerContainer
+from .OPCUAServerContainer import OPCUAServerContainer
class SingleNodeDockerCluster(Cluster):
@@ -54,7 +55,7 @@ class SingleNodeDockerCluster(Cluster):
logging.debug('Creating network: %s', net_name)
return docker.from_env().networks.create(net_name)
- def acquire_container(self, name, engine='minifi-cpp'):
+ def acquire_container(self, name, engine='minifi-cpp', command=None):
if name is not None and name in self.containers:
return self.containers[name]
@@ -63,23 +64,25 @@ class SingleNodeDockerCluster(Cluster):
logging.info('Container name was not provided; using generated name \'%s\'', name)
if engine == 'nifi':
- return self.containers.setdefault(name, NifiContainer(self.data_directories["nifi_config_dir"], name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, NifiContainer(self.data_directories["nifi_config_dir"], name, self.vols, self.network, self.image_store, command))
elif engine == 'minifi-cpp':
- return self.containers.setdefault(name, MinifiContainer(self.data_directories["minifi_config_dir"], name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, MinifiContainer(self.data_directories["minifi_config_dir"], name, self.vols, self.network, self.image_store, command))
elif engine == 'kafka-broker':
if 'zookeeper' not in self.containers:
- self.containers.setdefault('zookeeper', ZookeeperContainer('zookeeper', self.vols, self.network, self.image_store))
- return self.containers.setdefault(name, KafkaBrokerContainer(name, self.vols, self.network, self.image_store))
+ self.containers.setdefault('zookeeper', ZookeeperContainer('zookeeper', self.vols, self.network, self.image_store, command))
+ return self.containers.setdefault(name, KafkaBrokerContainer(name, self.vols, self.network, self.image_store, command))
elif engine == 'http-proxy':
- return self.containers.setdefault(name, HttpProxyContainer(name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, HttpProxyContainer(name, self.vols, self.network, self.image_store, command))
elif engine == 's3-server':
- return self.containers.setdefault(name, S3ServerContainer(name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, S3ServerContainer(name, self.vols, self.network, self.image_store, command))
elif engine == 'azure-storage-server':
- return self.containers.setdefault(name, AzureStorageServerContainer(name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, AzureStorageServerContainer(name, self.vols, self.network, self.image_store, command))
elif engine == 'postgresql-server':
- return self.containers.setdefault(name, PostgreSQLServerContainer(name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, PostgreSQLServerContainer(name, self.vols, self.network, self.image_store, command))
elif engine == 'mqtt-broker':
- return self.containers.setdefault(name, MqttBrokerContainer(name, self.vols, self.network, self.image_store))
+ return self.containers.setdefault(name, MqttBrokerContainer(name, self.vols, self.network, self.image_store, command))
+ elif engine == 'opcua-server':
+ return self.containers.setdefault(name, OPCUAServerContainer(name, self.vols, self.network, self.image_store, command))
else:
raise Exception('invalid flow engine: \'%s\'' % engine)
diff --git a/docker/test/integration/minifi/core/ZookeeperContainer.py b/docker/test/integration/minifi/core/ZookeeperContainer.py
index 09ec31b..107339e 100644
--- a/docker/test/integration/minifi/core/ZookeeperContainer.py
+++ b/docker/test/integration/minifi/core/ZookeeperContainer.py
@@ -3,8 +3,8 @@ from .Container import Container
class ZookeeperContainer(Container):
- def __init__(self, name, vols, network, image_store):
- super().__init__(name, 'zookeeper', vols, network, image_store)
+ def __init__(self, name, vols, network, image_store, command):
+ super().__init__(name, 'zookeeper', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
return "binding to port"
@@ -19,5 +19,6 @@ class ZookeeperContainer(Container):
detach=True,
name='zookeeper',
network=self.network.name,
- ports={'2181/tcp': 2181})
+ ports={'2181/tcp': 2181},
+ entrypoint=self.command)
logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/minifi/processors/FetchOPCProcessor.py b/docker/test/integration/minifi/processors/FetchOPCProcessor.py
new file mode 100644
index 0000000..cfa9b76
--- /dev/null
+++ b/docker/test/integration/minifi/processors/FetchOPCProcessor.py
@@ -0,0 +1,8 @@
+from ..core.Processor import Processor
+
+
+class FetchOPCProcessor(Processor):
+ def __init__(self):
+ super(FetchOPCProcessor, self).__init__(
+ 'FetchOPCProcessor',
+ auto_terminate=['success', 'failure'])
diff --git a/docker/test/integration/minifi/processors/PutOPCProcessor.py b/docker/test/integration/minifi/processors/PutOPCProcessor.py
new file mode 100644
index 0000000..47950e7
--- /dev/null
+++ b/docker/test/integration/minifi/processors/PutOPCProcessor.py
@@ -0,0 +1,9 @@
+from ..core.Processor import Processor
+
+
+class PutOPCProcessor(Processor):
+ def __init__(self, schedule={'scheduling strategy': 'EVENT_DRIVEN'}):
+ super(PutOPCProcessor, self).__init__(
+ 'PutOPCProcessor',
+ auto_terminate=['success', 'failure'],
+ schedule=schedule)
diff --git a/docker/test/integration/resources/opcua/opcua_client_cert.der b/docker/test/integration/resources/opcua/opcua_client_cert.der
new file mode 100644
index 0000000..44be253
Binary files /dev/null and b/docker/test/integration/resources/opcua/opcua_client_cert.der differ
diff --git a/docker/test/integration/resources/opcua/opcua_client_key.der b/docker/test/integration/resources/opcua/opcua_client_key.der
new file mode 100644
index 0000000..91b30bf
Binary files /dev/null and b/docker/test/integration/resources/opcua/opcua_client_key.der differ
diff --git a/docker/test/integration/steps/steps.py b/docker/test/integration/steps/steps.py
index c2fef16..f14dfe0 100644
--- a/docker/test/integration/steps/steps.py
+++ b/docker/test/integration/steps/steps.py
@@ -63,19 +63,31 @@ def step_impl(context, processor_type, property, property_value, processor_name)
format(processor_type=processor_type, property=property, property_value=property_value, minifi_container_name="minifi-cpp-flow", processor_name=processor_name))
-@given("a {processor_type} processor in the \"{minifi_container_name}\" flow")
-@given("a {processor_type} processor in a \"{minifi_container_name}\" flow")
-@given("a {processor_type} processor set up in a \"{minifi_container_name}\" flow")
-def step_impl(context, processor_type, minifi_container_name):
+@given("a {processor_type} processor with the name \"{processor_name}\" in the \"{minifi_container_name}\" flow")
+def step_impl(context, processor_type, processor_name, minifi_container_name):
container = context.test.acquire_container(minifi_container_name)
processor = locate("minifi.processors." + processor_type + "." + processor_type)()
- processor.set_name(processor_type)
+ processor.set_name(processor_name)
context.test.add_node(processor)
# Assume that the first node declared is primary unless specified otherwise
if not container.get_start_nodes():
container.add_start_node(processor)
+@given("a {processor_type} processor with the name \"{processor_name}\"")
+def step_impl(context, processor_type, processor_name):
+ context.execute_steps("given a {processor_type} processor with the name \"{processor_name}\" in the \"{minifi_container_name}\" flow".
+ format(processor_type=processor_type, processor_name=processor_name, minifi_container_name="minifi-cpp-flow"))
+
+
+@given("a {processor_type} processor in the \"{minifi_container_name}\" flow")
+@given("a {processor_type} processor in a \"{minifi_container_name}\" flow")
+@given("a {processor_type} processor set up in a \"{minifi_container_name}\" flow")
+def step_impl(context, processor_type, minifi_container_name):
+ context.execute_steps("given a {processor_type} processor with the name \"{processor_name}\" in the \"{minifi_container_name}\" flow".
+ format(processor_type=processor_type, processor_name=processor_type, minifi_container_name=minifi_container_name))
+
+
@given("a {processor_type} processor")
@given("a {processor_type} processor set up to communicate with an s3 server")
@given("a {processor_type} processor set up to communicate with the same s3 server")
@@ -344,6 +356,17 @@ def step_impl(context):
context.test.acquire_container("postgresql-server", "postgresql-server")
+# OPC UA
+@given("an OPC UA server is set up")
+def step_impl(context):
+ context.test.acquire_container("opcua-server", "opcua-server")
+
+
+@given("an OPC UA server is set up with access control")
+def step_impl(context):
+ context.test.acquire_container("opcua-server", "opcua-server", ["/opt/open62541/examples/access_control_server"])
+
+
@when("the MiNiFi instance starts up")
@when("both instances start up")
@when("all instances start up")
@@ -561,6 +584,11 @@ def step_impl(context, regex, duration):
context.test.check_minifi_log_matches_regex(regex, timeparse(duration))
+@then("the OPC UA server logs contain the following message: \"{log_message}\" in less than {duration}")
+def step_impl(context, log_message, duration):
+ context.test.check_container_log_contents("opcua-server", log_message, timeparse(duration))
+
+
# MQTT
@then("the MQTT broker has a log line matching \"{log_pattern}\"")
def step_impl(context, log_pattern):
diff --git a/extensions/opc/include/fetchopc.h b/extensions/opc/include/fetchopc.h
index 5652dd2..64bc443 100644
--- a/extensions/opc/include/fetchopc.h
+++ b/extensions/opc/include/fetchopc.h
@@ -91,6 +91,10 @@ class FetchOPCProcessor : public BaseOPCProcessor {
bool lazy_mode_;
private:
+ core::annotation::Input getInputRequirement() const override {
+ return core::annotation::Input::INPUT_FORBIDDEN;
+ }
+
std::vector<UA_NodeId> translatedNodeIDs_; // Only used when user provides path, path->nodeid translation is only done once
std::unordered_map<std::string, std::string> node_timestamp_; // Key = Full path, Value = Timestamp
};
diff --git a/extensions/opc/include/opc.h b/extensions/opc/include/opc.h
index 9646f3d..6472846 100644
--- a/extensions/opc/include/opc.h
+++ b/extensions/opc/include/opc.h
@@ -66,7 +66,7 @@ class Client {
UA_StatusCode update_node(const UA_NodeId nodeId, T value);
template<typename T>
- UA_StatusCode add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, T value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
+ UA_StatusCode add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, T value, UA_NodeId *receivedNodeId);
static std::unique_ptr<Client> createClient(std::shared_ptr<core::logging::Logger> logger, const std::string& applicationURI,
const std::vector<char>& certBuffer, const std::vector<char>& keyBuffer,
@@ -125,8 +125,6 @@ static std::map<std::string, OPCNodeDataType> StringToOPCDataTypeMap = {{"Int64
{"UInt32", OPCNodeDataType::UInt32}, {"Boolean", OPCNodeDataType::Boolean}, {"Float", OPCNodeDataType::Float},
{"Double", OPCNodeDataType::Double}, {"String", OPCNodeDataType::String}};
-int32_t OPCNodeDataTypeToTypeID(OPCNodeDataType dt);
-
std::string nodeValue2String(const NodeData& nd);
std::string OPCDateTime2String(UA_DateTime raw_date);
diff --git a/extensions/opc/include/putopc.h b/extensions/opc/include/putopc.h
index a18a583..4cb9707 100644
--- a/extensions/opc/include/putopc.h
+++ b/extensions/opc/include/putopc.h
@@ -80,6 +80,10 @@ class PutOPCProcessor : public BaseOPCProcessor {
std::shared_ptr<core::logging::Logger> logger_;
};
+ core::annotation::Input getInputRequirement() const override {
+ return core::annotation::Input::INPUT_REQUIRED;
+ }
+
std::string nodeID_;
int32_t nameSpaceIdx_;
opc::OPCNodeIDType idType_;
diff --git a/extensions/opc/src/opc.cpp b/extensions/opc/src/opc.cpp
index a85d36b..050246f 100644
--- a/extensions/opc/src/opc.cpp
+++ b/extensions/opc/src/opc.cpp
@@ -32,7 +32,6 @@
#include "open62541/client_highlevel.h"
#include "open62541/client_config_default.h"
-
namespace org {
namespace apache {
namespace nifi {
@@ -177,7 +176,10 @@ Client::~Client() {
if (client_ == nullptr) {
return;
}
- if (UA_Client_getState(client_) != UA_CLIENTSTATE_DISCONNECTED) {
+
+ UA_SecureChannelState channel_state;
+ UA_Client_getState(client_, &channel_state, nullptr, nullptr);
+ if (channel_state != UA_SECURECHANNELSTATE_CLOSED) {
auto sc = UA_Client_disconnect(client_);
if (sc != UA_STATUSCODE_GOOD) {
logger_->log_warn("Failed to disconnect OPC client: %s", UA_StatusCode_name(sc));
@@ -190,14 +192,17 @@ bool Client::isConnected() {
if (!client_) {
return false;
}
- return UA_Client_getState(client_) != UA_CLIENTSTATE_DISCONNECTED;
+
+ UA_SessionState session_state;
+ UA_Client_getState(client_, nullptr, &session_state, nullptr);
+ return session_state == UA_SESSIONSTATE_ACTIVATED;
}
UA_StatusCode Client::connect(const std::string& url, const std::string& username, const std::string& password) {
if (username.empty()) {
return UA_Client_connect(client_, url.c_str());
} else {
- return UA_Client_connect_username(client_, url.c_str(), username.c_str(), password.c_str());
+ return UA_Client_connectUsername(client_, url.c_str(), username.c_str(), password.c_str());
}
}
@@ -241,7 +246,7 @@ NodeData Client::getNodeData(const UA_ReferenceDescription *ref, const std::stri
auto server_timestamp = OPCDateTime2String(dv->serverTimestamp);
auto source_timestamp = OPCDateTime2String(dv->sourceTimestamp);
nodedata.attributes["Sourcetimestamp"] = source_timestamp;
- UA_ReadResponse_deleteMembers(&response);
+ UA_ReadResponse_clear(&response);
nodedata.dataTypeID = var->type->typeIndex;
nodedata.addVariant(var);
@@ -303,10 +308,10 @@ void Client::traverse(UA_NodeId nodeId, std::function<nodeFoundCallBackFunc> cb,
UA_BrowseResponse bResp = UA_Client_Service_browse(client_, bReq);
const auto guard = gsl::finally([&bResp]() {
- UA_BrowseResponse_deleteMembers(&bResp);
+ UA_BrowseResponse_clear(&bResp);
});
- UA_BrowseRequest_deleteMembers(&bReq);
+ UA_BrowseRequest_clear(&bReq);
for (size_t i = 0; i < bResp.resultsSize; ++i) {
for (size_t j = 0; j < bResp.results[i].referencesSize; ++j) {
@@ -364,7 +369,7 @@ UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& pa
UA_TranslateBrowsePathsToNodeIdsResponse response = UA_Client_Service_translateBrowsePathsToNodeIds(client_, request);
const auto guard = gsl::finally([&browsePath]() {
- UA_BrowsePath_deleteMembers(&browsePath);
+ UA_BrowsePath_clear(&browsePath);
});
if (response.resultsSize < 1) {
@@ -385,7 +390,7 @@ UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& pa
}
}
- UA_TranslateBrowsePathsToNodeIdsResponse_deleteMembers(&response);
+ UA_TranslateBrowsePathsToNodeIdsResponse_clear(&response);
if (foundData) {
logger->log_debug("Found %lu nodes for path %s", foundNodeIDs.size(), path.c_str());
@@ -397,7 +402,7 @@ UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& pa
}
template<typename T>
-UA_StatusCode Client::add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, T value, OPCNodeDataType dt, UA_NodeId *receivedNodeId) {
+UA_StatusCode Client::add_node(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, T value, UA_NodeId *receivedNodeId) {
UA_VariableAttributes attr = UA_VariableAttributes_default;
add_value_to_variant(&attr.value, value);
char local[6] = "en-US";
@@ -405,7 +410,7 @@ UA_StatusCode Client::add_node(const UA_NodeId parentNodeId, const UA_NodeId tar
UA_StatusCode sc = UA_Client_addVariableNode(client_,
targetNodeId,
parentNodeId,
- UA_NODEID_NUMERIC(0, OPCNodeDataTypeToTypeID(dt)),
+ UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES),
UA_QUALIFIEDNAME(1, const_cast<char*>(browseName.c_str())),
UA_NODEID_NULL,
attr, receivedNodeId);
@@ -443,40 +448,17 @@ template UA_StatusCode Client::update_node<bool>(const UA_NodeId nodeId, bool va
template UA_StatusCode Client::update_node<const char *>(const UA_NodeId nodeId, const char * value);
template UA_StatusCode Client::update_node<std::string>(const UA_NodeId nodeId, std::string value);
-template UA_StatusCode Client::add_node<int64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, int64_t value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<uint64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, uint64_t value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<int32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, int32_t value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<uint32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, uint32_t value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<float>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, float value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<double>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, double value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-template UA_StatusCode Client::add_node<bool>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, bool value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<int64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, int64_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<uint64_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, uint64_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<int32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, int32_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<uint32_t>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, uint32_t value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<float>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, float value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<double>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, double value, UA_NodeId *receivedNodeId);
+template UA_StatusCode Client::add_node<bool>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName, bool value, UA_NodeId *receivedNodeId);
template UA_StatusCode Client::add_node<const char *>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName,
- const char * value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
+ const char * value, UA_NodeId *receivedNodeId);
template UA_StatusCode Client::add_node<std::string>(const UA_NodeId parentNodeId, const UA_NodeId targetNodeId, std::string browseName,
- std::string value, OPCNodeDataType dt, UA_NodeId *receivedNodeId);
-
-int32_t OPCNodeDataTypeToTypeID(OPCNodeDataType dt) {
- switch (dt) {
- case OPCNodeDataType::Boolean:
- return UA_NS0ID_BOOLEAN;
- case OPCNodeDataType::Int32:
- return UA_NS0ID_INT32;
- case OPCNodeDataType::UInt32:
- return UA_NS0ID_UINT32;
- case OPCNodeDataType::Int64:
- return UA_NS0ID_INT64;
- case OPCNodeDataType::UInt64:
- return UA_NS0ID_UINT64;
- case OPCNodeDataType::Float:
- return UA_NS0ID_FLOAT;
- case OPCNodeDataType::Double:
- return UA_NS0ID_DOUBLE;
- case OPCNodeDataType::String:
- return UA_NS0ID_STRING;
- default:
- throw OPCException(GENERAL_EXCEPTION, "Data type is not supported");
- }
-}
+ std::string value, UA_NodeId *receivedNodeId);
std::string nodeValue2String(const NodeData& nd) {
std::string ret_val;
diff --git a/extensions/opc/src/opcbase.cpp b/extensions/opc/src/opcbase.cpp
index e675fea..03b18d7 100644
--- a/extensions/opc/src/opcbase.cpp
+++ b/extensions/opc/src/opcbase.cpp
@@ -44,14 +44,13 @@ namespace processors {
->withDescription("Application URI of the client in the format 'urn:unconfigured:application'. "
"Mandatory, if using Secure Channel and must match the URI included in the certificate's Subject Alternative Names.")->build());
-
core::Property BaseOPCProcessor::Username(
core::PropertyBuilder::createProperty("Username")
->withDescription("Username to log in with.")->build());
core::Property BaseOPCProcessor::Password(
core::PropertyBuilder::createProperty("Password")
- ->withDescription("Password to log in with. Providing this requires cert and key to be provided as well, credentials are always sent encrypted.")->build());
+ ->withDescription("Password to log in with.")->build());
core::Property BaseOPCProcessor::CertificatePath(
core::PropertyBuilder::createProperty("Certificate path")
@@ -84,45 +83,41 @@ namespace processors {
auto certificatePathRes = context->getProperty(CertificatePath.getName(), certpath_);
auto keyPathRes = context->getProperty(KeyPath.getName(), keypath_);
- auto trustedPathRes = context->getProperty(TrustedPath.getName(), trustpath_);
- if (certificatePathRes != keyPathRes || keyPathRes != trustedPathRes) {
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, "All or none of Certificate path, Key path and Trusted server certificate path should be provided!");
+ context->getProperty(TrustedPath.getName(), trustpath_);
+ if (certificatePathRes != keyPathRes) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "All or none of Certificate path and Key path should be provided!");
}
- if (!password_.empty() && (certpath_.empty() || keypath_.empty() || trustpath_.empty() || applicationURI_.empty())) {
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Certificate path, Key path, Trusted server certificate path and Application URI must be provided in case Password is provided!");
+ if (certpath_.empty()) {
+ return;
+ }
+ if (applicationURI_.empty()) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Application URI must be provided if Certificate path is provided!");
}
- if (!certpath_.empty()) {
- if (applicationURI_.empty()) {
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Application URI must be provided if Certificate path is provided!");
- }
+ std::ifstream input_cert(certpath_, std::ios::binary);
+ if (input_cert.good()) {
+ certBuffer_ = std::vector<char>(std::istreambuf_iterator<char>(input_cert), {});
+ }
+ std::ifstream input_key(keypath_, std::ios::binary);
+ if (input_key.good()) {
+ keyBuffer_ = std::vector<char>(std::istreambuf_iterator<char>(input_key), {});
+ }
- std::ifstream input_cert(certpath_, std::ios::binary);
- if (input_cert.good()) {
- certBuffer_ = std::vector<char>(std::istreambuf_iterator<char>(input_cert), {});
- }
- std::ifstream input_key(keypath_, std::ios::binary);
- if (input_key.good()) {
- keyBuffer_ = std::vector<char>(std::istreambuf_iterator<char>(input_key), {});
- }
+ if (certBuffer_.empty()) {
+ auto error_msg = utils::StringUtils::join_pack("Failed to load cert from path: ", certpath_);
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg);
+ }
+ if (keyBuffer_.empty()) {
+ auto error_msg = utils::StringUtils::join_pack("Failed to load key from path: ", keypath_);
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg);
+ }
- trustBuffers_.emplace_back();
+ if (!trustpath_.empty()) {
std::ifstream input_trust(trustpath_, std::ios::binary);
if (input_trust.good()) {
- trustBuffers_[0] = std::vector<char>(std::istreambuf_iterator<char>(input_trust), {});
- }
-
- if (certBuffer_.empty()) {
- auto error_msg = utils::StringUtils::join_pack("Failed to load cert from path: ", certpath_);
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg);
- }
- if (keyBuffer_.empty()) {
- auto error_msg = utils::StringUtils::join_pack("Failed to load key from path: ", keypath_);
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg);
- }
-
- if (trustBuffers_[0].empty()) {
+ trustBuffers_.push_back(std::vector<char>(std::istreambuf_iterator<char>(input_trust), {}));
+ } else {
auto error_msg = utils::StringUtils::join_pack("Failed to load trusted server certs from path: ", trustpath_);
throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg);
}
diff --git a/extensions/opc/src/putopc.cpp b/extensions/opc/src/putopc.cpp
index 8e4368a..02e2ddd 100644
--- a/extensions/opc/src/putopc.cpp
+++ b/extensions/opc/src/putopc.cpp
@@ -353,28 +353,28 @@ namespace processors {
switch (nodeDataType_) {
case opc::OPCNodeDataType::Int64: {
int64_t value = std::stoll(contentstr);
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, &resultnode);
break;
}
case opc::OPCNodeDataType::UInt64: {
uint64_t value = std::stoull(contentstr);
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, &resultnode);
break;
}
case opc::OPCNodeDataType::Int32: {
int32_t value = std::stoi(contentstr);
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, &resultnode);
break;
}
case opc::OPCNodeDataType::UInt32: {
uint32_t value = std::stoul(contentstr);
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, &resultnode);
break;
}
case opc::OPCNodeDataType::Boolean: {
const auto contentstr_parsed = utils::StringUtils::toBool(contentstr);
if (contentstr_parsed) {
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, contentstr_parsed.value(), nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, contentstr_parsed.value(), &resultnode);
} else {
throw opc::OPCException(GENERAL_EXCEPTION, "Content cannot be converted to bool");
}
@@ -382,16 +382,16 @@ namespace processors {
}
case opc::OPCNodeDataType::Float: {
float value = std::stof(contentstr);
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, &resultnode);
break;
}
case opc::OPCNodeDataType::Double: {
double value = std::stod(contentstr);
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, value, &resultnode);
break;
}
case opc::OPCNodeDataType::String: {
- sc = connection_->add_node(parentNodeID_, targetnode, browsename, contentstr, nodeDataType_, &resultnode);
+ sc = connection_->add_node(parentNodeID_, targetnode, browsename, contentstr, &resultnode);
break;
}
default:
diff --git a/thirdparty/open62541/open62541.patch b/thirdparty/open62541/open62541.patch
index 2d36e2a..17a5ab1 100644
--- a/thirdparty/open62541/open62541.patch
+++ b/thirdparty/open62541/open62541.patch
@@ -1,5 +1,5 @@
diff --git a/CMakeLists.txt b/CMakeLists.txt
-index d426e1da..5f1a4044 100644
+index 9184b943..44ae77cf 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -7,7 +7,7 @@ endif()
@@ -11,11 +11,20 @@ index d426e1da..5f1a4044 100644
find_package(PythonInterp REQUIRED)
find_package(Git)
include(AssignSourceGroup)
-@@ -416,17 +416,17 @@ if(NOT UA_COMPILE_AS_CXX AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID
+@@ -526,7 +526,7 @@ if(NOT UA_FORCE_CPP AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID}" ST
+ check_add_cc_flag("-Wall") # Warnings
+ check_add_cc_flag("-Wextra") # More warnings
+ check_add_cc_flag("-Wpedantic") # Standard compliance
+- check_add_cc_flag("-Werror") # All warnings are errors
++ # check_add_cc_flag("-Werror") # All warnings are errors
+
+ check_add_cc_flag("-Wno-static-in-inline") # Clang doesn't like the use of static inline methods inside static inline methods
+ check_add_cc_flag("-Wno-overlength-strings") # May happen in the nodeset compiler when complex values are directly encoded
+@@ -576,17 +576,17 @@ if(NOT UA_FORCE_CPP AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID}" ST
# IPO requires too much memory for unit tests
# GCC docu recommends to compile all files with the same options, therefore ignore it completely
-- if(NOT UA_BUILD_UNIT_TESTS)
+- if(NOT UA_BUILD_UNIT_TESTS AND NOT DEFINED CMAKE_INTERPROCEDURAL_OPTIMIZATION)
- # needed to check if IPO is supported (check needs cmake > 3.9)
- if("${CMAKE_VERSION}" VERSION_GREATER 3.9)
- cmake_policy(SET CMP0069 NEW) # needed as long as required cmake < 3.9
@@ -26,30 +35,17 @@ index d426e1da..5f1a4044 100644
- endif()
- endif()
- endif()
-+ #if(NOT UA_BUILD_UNIT_TESTS)
-+ # # needed to check if IPO is supported (check needs cmake > 3.9)
-+ # if("${CMAKE_VERSION}" VERSION_GREATER 3.9)
-+ # cmake_policy(SET CMP0069 NEW) # needed as long as required cmake < 3.9
-+ # include(CheckIPOSupported)
-+ # check_ipo_supported(RESULT CC_HAS_IPO) # Inter Procedural Optimization / Link Time Optimization (should be same as -flto)
-+ # if(CC_HAS_IPO)
-+ # set(CMAKE_INTERPROCEDURAL_OPTIMIZATION ON)
-+ # endif()
-+ # endif()
-+ #endif()
++ # if(NOT UA_BUILD_UNIT_TESTS AND NOT DEFINED CMAKE_INTERPROCEDURAL_OPTIMIZATION)
++ # # needed to check if IPO is supported (check needs cmake > 3.9)
++ # if("${CMAKE_VERSION}" VERSION_GREATER 3.9)
++ # cmake_policy(SET CMP0069 NEW) # needed as long as required cmake < 3.9
++ # include(CheckIPOSupported)
++ # check_ipo_supported(RESULT CC_HAS_IPO) # Inter Procedural Optimization / Link Time Optimization (should be same as -flto)
++ # if(CC_HAS_IPO)
++ # set(CMAKE_INTERPROCEDURAL_OPTIMIZATION ON)
++ # endif()
++ # endif()
++ # endif()
endif()
if(UA_ENABLE_AMALGAMATION)
-diff --git a/CMakeLists.txt b/CMakeLists.txt
-index 94ceb127..5d0ea914 100644
---- a/CMakeLists.txt
-+++ b/CMakeLists.txt
-@@ -405,7 +405,7 @@ if(NOT UA_COMPILE_AS_CXX AND (CMAKE_COMPILER_IS_GNUCC OR "x${CMAKE_C_COMPILER_ID
- -fno-strict-aliasing # fewer compiler assumptions about pointer types
- -fexceptions # recommended for multi-threaded C code, also in combination with C++ code
- )
-- set(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -Werror")
-+ # set(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -Werror")
-
- if (NOT MINGW)
- if(UA_ENABLE_HARDENING)
[nifi-minifi-cpp] 04/05: MINIFICPP-1673 ProcessSession write/append should only operate on new/modified flowfiles
Posted by sz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 85c4f1cf0d2a44f3c71b26d612ec1e4ee430b053
Author: Martin Zink <ma...@apache.org>
AuthorDate: Tue Jan 4 16:20:53 2022 +0100
MINIFICPP-1673 ProcessSession write/append should only operate on new/modified flowfiles
Closes #1226
Signed-off-by: Marton Szasz <sz...@apache.org>
---
extensions/http-curl/tests/unit/InvokeHTTPTests.cpp | 20 --------------------
libminifi/src/core/ProcessSession.cpp | 10 ++++++++++
2 files changed, 10 insertions(+), 20 deletions(-)
diff --git a/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp b/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
index 00f76f5..ae00f85 100644
--- a/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
+++ b/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
@@ -109,20 +109,6 @@ TEST_CASE("HTTPTestsWithNoResourceClaimPOST", "[httptest1]") {
REQUIRE(LogTestController::getInstance().contains("Exiting because method is POST"));
}
-class CallBack : public minifi::OutputStreamCallback {
- public:
- CallBack() {
- }
- virtual ~CallBack() {
- }
- virtual int64_t process(const std::shared_ptr<minifi::io::BaseStream>& stream) {
- // leaving the typo for posterity sake
- std::string st = "we're gnna write some test stuff";
- const auto write_ret = stream->write(reinterpret_cast<const uint8_t*>(st.c_str()), st.length());
- return minifi::io::isError(write_ret) ? -1 : gsl::narrow<int64_t>(write_ret);
- }
-};
-
TEST_CASE("HTTPTestsWithResourceClaimPOST", "[httptest1]") {
TestController testController;
@@ -181,12 +167,6 @@ TEST_CASE("HTTPTestsWithResourceClaimPOST", "[httptest1]") {
std::shared_ptr<core::FlowFile> record;
- CallBack callback;
-
- auto flow = std::make_shared<minifi::FlowFileRecord>();
- flow->setAttribute("testy", "test");
- session2->write(flow, &callback);
-
invokehttp->incrementActiveTasks();
invokehttp->setScheduledState(core::ScheduledState::RUNNING);
std::shared_ptr<core::ProcessSessionFactory> factory2 = std::make_shared<core::ProcessSessionFactory>(context2);
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index 4d318e3..6c6136f 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -229,6 +229,11 @@ void ProcessSession::transfer(const std::shared_ptr<core::FlowFile> &flow, Relat
}
void ProcessSession::write(const std::shared_ptr<core::FlowFile> &flow, OutputStreamCallback *callback) {
+ auto flow_file_equality_checker = [&flow](const auto& flow_file) { return flow == flow_file; };
+ gsl_ExpectsAudit(_updatedFlowFiles.contains(flow->getUUID())
+ || _addedFlowFiles.contains(flow->getUUID())
+ || std::any_of(_clonedFlowFiles.begin(), _clonedFlowFiles.end(), flow_file_equality_checker));
+
std::shared_ptr<ResourceClaim> claim = content_session_->create();
try {
@@ -272,6 +277,11 @@ void ProcessSession::writeBuffer(const std::shared_ptr<core::FlowFile>& flow_fil
}
void ProcessSession::append(const std::shared_ptr<core::FlowFile> &flow, OutputStreamCallback *callback) {
+ auto flow_file_equality_checker = [&flow](const auto& flow_file) { return flow == flow_file; };
+ gsl_ExpectsAudit(_updatedFlowFiles.contains(flow->getUUID())
+ || _addedFlowFiles.contains(flow->getUUID())
+ || std::any_of(_clonedFlowFiles.begin(), _clonedFlowFiles.end(), flow_file_equality_checker));
+
std::shared_ptr<ResourceClaim> claim = flow->getResourceClaim();
if (!claim) {
// No existed claim for append, we need to create new claim