You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ph...@apache.org on 2017/11/21 13:05:07 UTC
nifi-minifi-cpp git commit: MINIFICPP-308 Ensure that all core
resources are deterministically released at the end of onTrigger calls so
that script engines using GC do not hold onto them and waste resources
Repository: nifi-minifi-cpp
Updated Branches:
refs/heads/master 8b7d76ade -> 5bd67470e
MINIFICPP-308 Ensure that all core resources are deterministically released at the end of onTrigger calls so that script engines using GC do not hold onto them and waste resources
This closes #190.
Signed-off-by: Marc Parisi <ph...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/5bd67470
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/5bd67470
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/5bd67470
Branch: refs/heads/master
Commit: 5bd67470e3922c0b51a1f60998c5b28fcbcdd363
Parents: 8b7d76a
Author: Andrew I. Christianson <an...@andyic.org>
Authored: Wed Nov 15 19:22:49 2017 -0500
Committer: Marc Parisi <ph...@apache.org>
Committed: Tue Nov 21 08:04:41 2017 -0500
----------------------------------------------------------------------
extensions/script/ExecuteScript.h | 6 +-
extensions/script/ScriptFlowFile.cpp | 20 +
extensions/script/ScriptFlowFile.h | 1 +
extensions/script/ScriptProcessContext.cpp | 4 +
extensions/script/ScriptProcessContext.h | 1 +
extensions/script/lua/LuaProcessSession.cpp | 82 ++-
extensions/script/lua/LuaProcessSession.h | 13 +-
extensions/script/lua/LuaScriptEngine.cpp | 2 +-
extensions/script/lua/LuaScriptEngine.h | 34 +-
extensions/script/python/PyProcessSession.cpp | 84 ++-
extensions/script/python/PyProcessSession.h | 11 +
extensions/script/python/PythonScriptEngine.h | 34 +-
.../test/script-tests/ExecuteScriptTests.cpp | 640 -------------------
13 files changed, 261 insertions(+), 671 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/ExecuteScript.h
----------------------------------------------------------------------
diff --git a/extensions/script/ExecuteScript.h b/extensions/script/ExecuteScript.h
index 441969a..fbfaa63 100644
--- a/extensions/script/ExecuteScript.h
+++ b/extensions/script/ExecuteScript.h
@@ -26,6 +26,7 @@
#include <core/Processor.h>
#include "ScriptEngine.h"
+#include "ScriptProcessContext.h"
namespace org {
namespace apache {
@@ -54,7 +55,8 @@ class ExecuteScript : public core::Processor {
void onTrigger(core::ProcessContext *context, core::ProcessSession *session) override {
logger_->log_error("onTrigger invocation with raw pointers is not implemented");
}
- void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) override;
+ void onTrigger(const std::shared_ptr<core::ProcessContext> &context,
+ const std::shared_ptr<core::ProcessSession> &session) override;
private:
std::shared_ptr<logging::Logger> logger_;
@@ -82,7 +84,7 @@ class ExecuteScript : public core::Processor {
const std::shared_ptr<core::ProcessContext> &context,
const std::shared_ptr<core::ProcessSession> &session) const {
auto typed_engine = std::static_pointer_cast<T>(engine);
- typed_engine->call("onTrigger", context, session);
+ typed_engine->onTrigger(context, session);
}
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/ScriptFlowFile.cpp
----------------------------------------------------------------------
diff --git a/extensions/script/ScriptFlowFile.cpp b/extensions/script/ScriptFlowFile.cpp
index 79ec073..a71f454 100644
--- a/extensions/script/ScriptFlowFile.cpp
+++ b/extensions/script/ScriptFlowFile.cpp
@@ -35,20 +35,36 @@ ScriptFlowFile::ScriptFlowFile(std::shared_ptr<core::FlowFile> flow_file)
}
std::string ScriptFlowFile::getAttribute(const std::string &key) {
+ if (!flow_file_) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
std::string value;
flow_file_->getAttribute(key, value);
return value;
}
bool ScriptFlowFile::addAttribute(const std::string &key, const std::string &value) {
+ if (!flow_file_) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
return flow_file_->addAttribute(key, value);
}
bool ScriptFlowFile::updateAttribute(std::string key, std::string value) {
+ if (!flow_file_) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
return flow_file_->updateAttribute(std::move(key), std::move(value));
}
bool ScriptFlowFile::removeAttribute(std::string key) {
+ if (!flow_file_) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
return flow_file_->removeAttribute(std::move(key));
}
@@ -56,6 +72,10 @@ std::shared_ptr<core::FlowFile> ScriptFlowFile::getFlowFile() {
return flow_file_;
}
+void ScriptFlowFile::releaseFlowFile() {
+ flow_file_.reset();
+}
+
} /* namespace script */
} /* namespace minifi */
} /* namespace nifi */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/ScriptFlowFile.h
----------------------------------------------------------------------
diff --git a/extensions/script/ScriptFlowFile.h b/extensions/script/ScriptFlowFile.h
index 707cf01..fbbae49 100644
--- a/extensions/script/ScriptFlowFile.h
+++ b/extensions/script/ScriptFlowFile.h
@@ -35,6 +35,7 @@ class ScriptFlowFile {
bool updateAttribute(std::string key, std::string value);
bool removeAttribute(std::string key);
std::shared_ptr<core::FlowFile> getFlowFile();
+ void releaseFlowFile();
private:
std::shared_ptr<core::FlowFile> flow_file_;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/ScriptProcessContext.cpp
----------------------------------------------------------------------
diff --git a/extensions/script/ScriptProcessContext.cpp b/extensions/script/ScriptProcessContext.cpp
index 76aa6b7..cff4b15 100644
--- a/extensions/script/ScriptProcessContext.cpp
+++ b/extensions/script/ScriptProcessContext.cpp
@@ -37,6 +37,10 @@ std::string ScriptProcessContext::getProperty(const std::string &name) {
return value;
}
+void ScriptProcessContext::releaseProcessContext() {
+ context_.reset();
+}
+
} /* namespace script */
} /* namespace minifi */
} /* namespace nifi */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/ScriptProcessContext.h
----------------------------------------------------------------------
diff --git a/extensions/script/ScriptProcessContext.h b/extensions/script/ScriptProcessContext.h
index 4f240b1..3a92c26 100644
--- a/extensions/script/ScriptProcessContext.h
+++ b/extensions/script/ScriptProcessContext.h
@@ -34,6 +34,7 @@ class ScriptProcessContext {
explicit ScriptProcessContext(std::shared_ptr<core::ProcessContext> context);
std::string getProperty(const std::string &name);
+ void releaseProcessContext();
private:
std::shared_ptr<core::ProcessContext> context_;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/lua/LuaProcessSession.cpp
----------------------------------------------------------------------
diff --git a/extensions/script/lua/LuaProcessSession.cpp b/extensions/script/lua/LuaProcessSession.cpp
index ce91c62..3841dd4 100644
--- a/extensions/script/lua/LuaProcessSession.cpp
+++ b/extensions/script/lua/LuaProcessSession.cpp
@@ -30,42 +30,104 @@ LuaProcessSession::LuaProcessSession(std::shared_ptr<core::ProcessSession> sessi
}
std::shared_ptr<script::ScriptFlowFile> LuaProcessSession::get() {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
auto flow_file = session_->get();
if (flow_file == nullptr) {
return nullptr;
}
- return std::make_shared<script::ScriptFlowFile>(flow_file);
+ auto result = std::make_shared<script::ScriptFlowFile>(flow_file);
+ flow_files_.push_back(result);
+
+ return result;
}
-void LuaProcessSession::transfer(const std::shared_ptr<script::ScriptFlowFile> &flow_file,
+void LuaProcessSession::transfer(const std::shared_ptr<script::ScriptFlowFile> &script_flow_file,
core::Relationship relationship) {
- session_->transfer(flow_file->getFlowFile(), relationship);
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto flow_file = script_flow_file->getFlowFile();
+
+ if (!flow_file) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
+ session_->transfer(flow_file, relationship);
}
-void LuaProcessSession::read(const std::shared_ptr<script::ScriptFlowFile> &flow_file,
+void LuaProcessSession::read(const std::shared_ptr<script::ScriptFlowFile> &script_flow_file,
sol::table input_stream_callback) {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto flow_file = script_flow_file->getFlowFile();
+
+ if (!flow_file) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
LuaInputStreamCallback lua_callback(input_stream_callback);
- session_->read(flow_file->getFlowFile(), &lua_callback);
+ session_->read(flow_file, &lua_callback);
}
-void LuaProcessSession::write(const std::shared_ptr<script::ScriptFlowFile> &flow_file,
+void LuaProcessSession::write(const std::shared_ptr<script::ScriptFlowFile> &script_flow_file,
sol::table output_stream_callback) {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto flow_file = script_flow_file->getFlowFile();
+
+ if (!flow_file) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
LuaOutputStreamCallback lua_callback(output_stream_callback);
- session_->write(flow_file->getFlowFile(), &lua_callback);
+ session_->write(flow_file, &lua_callback);
}
std::shared_ptr<script::ScriptFlowFile> LuaProcessSession::create() {
- return std::make_shared<script::ScriptFlowFile>(session_->create());
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto result = std::make_shared<script::ScriptFlowFile>(session_->create());
+ flow_files_.push_back(result);
+ return result;
}
std::shared_ptr<script::ScriptFlowFile> LuaProcessSession::create(const std::shared_ptr<script::ScriptFlowFile> &flow_file) {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ std::shared_ptr<script::ScriptFlowFile> result;
+
if (flow_file == nullptr) {
- return std::make_shared<script::ScriptFlowFile>(session_->create());
+ result = std::make_shared<script::ScriptFlowFile>(session_->create());
} else {
- return std::make_shared<script::ScriptFlowFile>(session_->create(flow_file->getFlowFile()));
+ result = std::make_shared<script::ScriptFlowFile>(session_->create(flow_file->getFlowFile()));
}
+
+ flow_files_.push_back(result);
+ return result;
+}
+
+void LuaProcessSession::releaseCoreResources() {
+ for (const auto &flow_file : flow_files_) {
+ if (flow_file) {
+ flow_file->releaseFlowFile();
+ }
+ }
+
+ session_.reset();
}
} /* namespace lua */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/lua/LuaProcessSession.h
----------------------------------------------------------------------
diff --git a/extensions/script/lua/LuaProcessSession.h b/extensions/script/lua/LuaProcessSession.h
index 4752d0d..535f91b 100644
--- a/extensions/script/lua/LuaProcessSession.h
+++ b/extensions/script/lua/LuaProcessSession.h
@@ -40,9 +40,19 @@ class LuaProcessSession {
std::shared_ptr<script::ScriptFlowFile> create();
std::shared_ptr<script::ScriptFlowFile> create(const std::shared_ptr<script::ScriptFlowFile> &flow_file);
void transfer(const std::shared_ptr<script::ScriptFlowFile> &flow_file, core::Relationship relationship);
- void read(const std::shared_ptr<script::ScriptFlowFile> &flow_file, sol::table input_stream_callback);
+ void read(const std::shared_ptr<script::ScriptFlowFile> &script_flow_file, sol::table input_stream_callback);
void write(const std::shared_ptr<script::ScriptFlowFile> &flow_file, sol::table output_stream_callback);
+ /**
+ * Sometimes we want to release shared pointers to core resources when
+ * we know they are no longer in need. This method is for those times.
+ *
+ * For example, we do not want to hold on to shared pointers to FlowFiles
+ * after an onTrigger call, because doing so can be very expensive in terms
+ * of repository resources.
+ */
+ void releaseCoreResources();
+
class LuaInputStreamCallback : public InputStreamCallback {
public:
explicit LuaInputStreamCallback(const sol::table &input_stream_callback) {
@@ -76,6 +86,7 @@ class LuaProcessSession {
};
private:
+ std::vector<std::shared_ptr<script::ScriptFlowFile>> flow_files_;
std::shared_ptr<core::ProcessSession> session_;
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/lua/LuaScriptEngine.cpp
----------------------------------------------------------------------
diff --git a/extensions/script/lua/LuaScriptEngine.cpp b/extensions/script/lua/LuaScriptEngine.cpp
index 9d1a7f5..9ba2631 100644
--- a/extensions/script/lua/LuaScriptEngine.cpp
+++ b/extensions/script/lua/LuaScriptEngine.cpp
@@ -51,7 +51,7 @@ LuaScriptEngine::LuaScriptEngine()
"write", &lua::LuaProcessSession::write,
"transfer", &lua::LuaProcessSession::transfer);
lua_.new_usertype<script::ScriptFlowFile>(
- "ProcessSession",
+ "FlowFile",
"getAttribute", &script::ScriptFlowFile::getAttribute,
"addAttribute", &script::ScriptFlowFile::addAttribute,
"removeAttribute", &script::ScriptFlowFile::removeAttribute,
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/lua/LuaScriptEngine.h
----------------------------------------------------------------------
diff --git a/extensions/script/lua/LuaScriptEngine.h b/extensions/script/lua/LuaScriptEngine.h
index ec40ddb..82f4ba7 100644
--- a/extensions/script/lua/LuaScriptEngine.h
+++ b/extensions/script/lua/LuaScriptEngine.h
@@ -51,6 +51,32 @@ class LuaScriptEngine : public script::ScriptEngine {
fn(convert(args)...);
}
+ class TriggerSession {
+ public:
+ TriggerSession(std::shared_ptr<script::ScriptProcessContext> script_context,
+ std::shared_ptr<lua::LuaProcessSession> lua_session)
+ : script_context_(std::move(script_context)),
+ lua_session_(std::move(lua_session)) {
+ }
+
+ ~TriggerSession() {
+ script_context_->releaseProcessContext();
+ lua_session_->releaseCoreResources();
+ }
+
+
+ private:
+ std::shared_ptr<script::ScriptProcessContext> script_context_;
+ std::shared_ptr<LuaProcessSession> lua_session_;
+ };
+
+ void onTrigger(const std::shared_ptr<core::ProcessContext> &context,
+ const std::shared_ptr<core::ProcessSession> &session) {
+ auto script_context = convert(context);
+ auto lua_session = convert(session);
+ TriggerSession trigger_session(script_context, lua_session);
+ call("onTrigger", script_context, lua_session);
+ }
template<typename T>
void bind(const std::string &name, const T &value) {
@@ -62,14 +88,14 @@ class LuaScriptEngine : public script::ScriptEngine {
return value;
}
- std::shared_ptr<LuaProcessSession> convert(const std::shared_ptr<core::ProcessSession> &session) {
- return std::make_shared<LuaProcessSession>(session);
- }
-
std::shared_ptr<script::ScriptProcessContext> convert(const std::shared_ptr<core::ProcessContext> &context) {
return std::make_shared<script::ScriptProcessContext>(context);
}
+ std::shared_ptr<LuaProcessSession> convert(const std::shared_ptr<core::ProcessSession> &session) {
+ return std::make_shared<LuaProcessSession>(session);
+ }
+
private:
sol::state lua_;
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/python/PyProcessSession.cpp
----------------------------------------------------------------------
diff --git a/extensions/script/python/PyProcessSession.cpp b/extensions/script/python/PyProcessSession.cpp
index 2a72ad2..2ccbc75 100644
--- a/extensions/script/python/PyProcessSession.cpp
+++ b/extensions/script/python/PyProcessSession.cpp
@@ -36,38 +36,104 @@ PyProcessSession::PyProcessSession(std::shared_ptr<core::ProcessSession> session
}
std::shared_ptr<script::ScriptFlowFile> PyProcessSession::get() {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
auto flow_file = session_->get();
if (flow_file == nullptr) {
return nullptr;
}
- return std::make_shared<script::ScriptFlowFile>(flow_file);
+ auto result = std::make_shared<script::ScriptFlowFile>(flow_file);
+ flow_files_.push_back(result);
+
+ return result;
}
-void PyProcessSession::transfer(std::shared_ptr<script::ScriptFlowFile> flow_file,
+void PyProcessSession::transfer(std::shared_ptr<script::ScriptFlowFile> script_flow_file,
core::Relationship relationship) {
- session_->transfer(flow_file->getFlowFile(), relationship);
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto flow_file = script_flow_file->getFlowFile();
+
+ if (!flow_file) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
+ session_->transfer(flow_file, relationship);
}
-void PyProcessSession::read(std::shared_ptr<script::ScriptFlowFile> flow_file,
+void PyProcessSession::read(std::shared_ptr<script::ScriptFlowFile> script_flow_file,
py::object input_stream_callback) {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto flow_file = script_flow_file->getFlowFile();
+
+ if (!flow_file) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
PyInputStreamCallback py_callback(input_stream_callback);
- session_->read(flow_file->getFlowFile(), &py_callback);
+ session_->read(flow_file, &py_callback);
}
-void PyProcessSession::write(std::shared_ptr<script::ScriptFlowFile> flow_file,
+void PyProcessSession::write(std::shared_ptr<script::ScriptFlowFile> script_flow_file,
py::object output_stream_callback) {
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto flow_file = script_flow_file->getFlowFile();
+
+ if (!flow_file) {
+ throw std::runtime_error("Access of FlowFile after it has been released");
+ }
+
PyOutputStreamCallback py_callback(output_stream_callback);
- session_->write(flow_file->getFlowFile(), &py_callback);
+ session_->write(flow_file, &py_callback);
}
std::shared_ptr<script::ScriptFlowFile> PyProcessSession::create() {
- return std::make_shared<script::ScriptFlowFile>(session_->create());
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ auto result = std::make_shared<script::ScriptFlowFile>(session_->create());
+ flow_files_.push_back(result);
+ return result;
}
std::shared_ptr<script::ScriptFlowFile> PyProcessSession::create(std::shared_ptr<script::ScriptFlowFile> flow_file) {
- return std::make_shared<script::ScriptFlowFile>(session_->create(flow_file->getFlowFile()));
+ if (!session_) {
+ throw std::runtime_error("Access of ProcessSession after it has been released");
+ }
+
+ std::shared_ptr<script::ScriptFlowFile> result;
+
+ if (flow_file == nullptr) {
+ result = std::make_shared<script::ScriptFlowFile>(session_->create());
+ } else {
+ result = std::make_shared<script::ScriptFlowFile>(session_->create(flow_file->getFlowFile()));
+ }
+
+ flow_files_.push_back(result);
+ return result;
+}
+
+void PyProcessSession::releaseCoreResources() {
+ for (const auto &flow_file : flow_files_) {
+ if (flow_file) {
+ flow_file->releaseFlowFile();
+ }
+ }
+
+ session_.reset();
}
} /* namespace python */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/python/PyProcessSession.h
----------------------------------------------------------------------
diff --git a/extensions/script/python/PyProcessSession.h b/extensions/script/python/PyProcessSession.h
index 3581866..c760acd 100644
--- a/extensions/script/python/PyProcessSession.h
+++ b/extensions/script/python/PyProcessSession.h
@@ -46,6 +46,16 @@ class PyProcessSession {
void read(std::shared_ptr<script::ScriptFlowFile> flow_file, py::object input_stream_callback);
void write(std::shared_ptr<script::ScriptFlowFile> flow_file, py::object output_stream_callback);
+ /**
+ * Sometimes we want to release shared pointers to core resources when
+ * we know they are no longer in need. This method is for those times.
+ *
+ * For example, we do not want to hold on to shared pointers to FlowFiles
+ * after an onTrigger call, because doing so can be very expensive in terms
+ * of repository resources.
+ */
+ void releaseCoreResources();
+
class __attribute__((visibility("default"))) PyInputStreamCallback : public InputStreamCallback {
public:
explicit PyInputStreamCallback(const py::object &input_stream_callback) {
@@ -77,6 +87,7 @@ class PyProcessSession {
};
private:
+ std::vector<std::shared_ptr<script::ScriptFlowFile>> flow_files_;
std::shared_ptr<core::ProcessSession> session_;
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/extensions/script/python/PythonScriptEngine.h
----------------------------------------------------------------------
diff --git a/extensions/script/python/PythonScriptEngine.h b/extensions/script/python/PythonScriptEngine.h
index 575b4db..280d20a 100644
--- a/extensions/script/python/PythonScriptEngine.h
+++ b/extensions/script/python/PythonScriptEngine.h
@@ -74,6 +74,32 @@ class __attribute__((visibility("default"))) PythonScriptEngine : public script:
(*bindings_)[fn_name.c_str()](convert(args)...);
}
+ class TriggerSession {
+ public:
+ TriggerSession(std::shared_ptr<script::ScriptProcessContext> script_context,
+ std::shared_ptr<PyProcessSession> py_session)
+ : script_context_(std::move(script_context)),
+ py_session_(std::move(py_session)) {
+ }
+
+ ~TriggerSession() {
+ script_context_->releaseProcessContext();
+ py_session_->releaseCoreResources();
+ }
+
+ private:
+ std::shared_ptr<script::ScriptProcessContext> script_context_;
+ std::shared_ptr<PyProcessSession> py_session_;
+ };
+
+ void onTrigger(const std::shared_ptr<core::ProcessContext> &context,
+ const std::shared_ptr<core::ProcessSession> &session) {
+ auto script_context = convertContext(context);
+ auto py_session = convertSession(session);
+ TriggerSession trigger_session(script_context, py_session);
+ call("onTrigger", script_context, py_session);
+ }
+
/**
* Binds an object into the scope of the python interpreter.
* @tparam T
@@ -92,12 +118,12 @@ class __attribute__((visibility("default"))) PythonScriptEngine : public script:
return py::cast(value);
}
- py::object convert(const std::shared_ptr<core::ProcessSession> &session) {
- return py::cast(std::make_shared<PyProcessSession>(session));
+ std::shared_ptr<PyProcessSession> convertSession(const std::shared_ptr<core::ProcessSession> &session) {
+ return std::make_shared<PyProcessSession>(session);
}
- py::object convert(const std::shared_ptr<core::ProcessContext> &context) {
- return py::cast(std::make_shared<script::ScriptProcessContext>(context));
+ std::shared_ptr<script::ScriptProcessContext> convertContext(const std::shared_ptr<core::ProcessContext> &context) {
+ return std::make_shared<script::ScriptProcessContext>(context);
}
private:
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/5bd67470/libminifi/test/script-tests/ExecuteScriptTests.cpp
----------------------------------------------------------------------
diff --git a/libminifi/test/script-tests/ExecuteScriptTests.cpp b/libminifi/test/script-tests/ExecuteScriptTests.cpp
deleted file mode 100644
index 9af7926..0000000
--- a/libminifi/test/script-tests/ExecuteScriptTests.cpp
+++ /dev/null
@@ -1,640 +0,0 @@
-/**
- *
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#define CATCH_CONFIG_MAIN
-
-#include <memory>
-#include <string>
-#include <set>
-
-#include "../TestBase.h"
-
-TEST_CASE("Test Creation of ExecuteScript", "[executescriptCreate]") { // NOLINT
- TestController testController;
- auto processor = std::make_shared<processors::ExecuteScript>("processorname");
- REQUIRE(processor->getName() == "processorname");
-}
-
-TEST_CASE("Python: Test Read File", "[executescriptPythonRead]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto putFile = plan->addProcessor("PutFile", "putFile", core::Relationship("success", "description"), true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- import codecs
-
- class ReadCallback(object):
- def process(self, input_stream):
- content = codecs.getreader('utf-8')(input_stream).read()
- log.info('file content: %s' % content)
- return len(content)
-
- def onTrigger(context, session):
- flow_file = session.get()
-
- if flow_file is not None:
- log.info('got flow file: %s' % flow_file.getAttribute('filename'))
- session.read(flow_file, ReadCallback())
- session.transfer(flow_file, REL_SUCCESS)
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- char putFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *putFileDir = testController.createTempDirectory(putFileDirFmt);
- plan->setProperty(putFile, processors::PutFile::Directory.getName(), putFileDir);
-
- testController.runSession(plan, false);
-
- std::set<provenance::ProvenanceEventRecord *> records = plan->getProvenanceRecords();
- std::shared_ptr<core::FlowFile> record = plan->getCurrentFlowFile();
- REQUIRE(record == nullptr);
- REQUIRE(records.empty());
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- records = plan->getProvenanceRecords();
- record = plan->getCurrentFlowFile();
- testController.runSession(plan, false);
-
- unlink(ss.str().c_str());
-
- REQUIRE(logTestController.contains("[info] file content: tempFile"));
-
- // Verify that file content was preserved
- REQUIRE(!std::ifstream(ss.str()).good());
- std::stringstream movedFile;
- movedFile << putFileDir << "/" << "tstFile.ext";
- REQUIRE(std::ifstream(movedFile.str()).good());
-
- file.open(movedFile.str(), std::ios::in);
- std::string contents((std::istreambuf_iterator<char>(file)),
- std::istreambuf_iterator<char>());
- REQUIRE("tempFile" == contents);
- file.close();
- logTestController.reset();
-}
-
-TEST_CASE("Python: Test Write File", "[executescriptPythonWrite]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto putFile = plan->addProcessor("PutFile", "putFile", core::Relationship("success", "description"), true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- class WriteCallback(object):
- def process(self, output_stream):
- new_content = 'hello 2'.encode('utf-8')
- output_stream.write(new_content)
- return len(new_content)
-
- def onTrigger(context, session):
- flow_file = session.get()
- if flow_file is not None:
- log.info('got flow file: %s' % flow_file.getAttribute('filename'))
- session.write(flow_file, WriteCallback())
- session.transfer(flow_file, REL_SUCCESS)
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- char putFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *putFileDir = testController.createTempDirectory(putFileDirFmt);
- plan->setProperty(putFile, processors::PutFile::Directory.getName(), putFileDir);
-
- testController.runSession(plan, false);
-
- std::set<provenance::ProvenanceEventRecord *> records = plan->getProvenanceRecords();
- std::shared_ptr<core::FlowFile> record = plan->getCurrentFlowFile();
- REQUIRE(record == nullptr);
- REQUIRE(records.empty());
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- records = plan->getProvenanceRecords();
- record = plan->getCurrentFlowFile();
- testController.runSession(plan, false);
-
- unlink(ss.str().c_str());
-
- // Verify new content was written
- REQUIRE(!std::ifstream(ss.str()).good());
- std::stringstream movedFile;
- movedFile << putFileDir << "/" << "tstFile.ext";
- REQUIRE(std::ifstream(movedFile.str()).good());
-
- file.open(movedFile.str(), std::ios::in);
- std::string contents((std::istreambuf_iterator<char>(file)),
- std::istreambuf_iterator<char>());
- REQUIRE("hello 2" == contents);
- file.close();
- logTestController.reset();
-}
-
-TEST_CASE("Python: Test Create", "[executescriptPythonCreate]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript");
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- def onTrigger(context, session):
- flow_file = session.create()
-
- if flow_file is not None:
- log.info('created flow file: %s' % flow_file.getAttribute('filename'))
- session.transfer(flow_file, REL_SUCCESS)
- )");
-
- plan->reset();
-
- testController.runSession(plan, false);
-
- REQUIRE(LogTestController::getInstance().contains("[info] created flow file:"));
-
- logTestController.reset();
-}
-
-TEST_CASE("Python: Test Update Attribute", "[executescriptPythonUpdateAttribute]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- def onTrigger(context, session):
- flow_file = session.get()
-
- if flow_file is not None:
- log.info('got flow file: %s' % flow_file.getAttribute('filename'))
- flow_file.addAttribute('test_attr', '1')
- attr = flow_file.getAttribute('test_attr')
- log.info('got flow file attr \'test_attr\': %s' % attr)
- flow_file.updateAttribute('test_attr', str(int(attr) + 1))
- session.transfer(flow_file, REL_SUCCESS)
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- REQUIRE(LogTestController::getInstance().contains("key:test_attr value:2"));
-
- logTestController.reset();
-}
-
-TEST_CASE("Python: Test Get Context Property", "[executescriptPythonGetContextProperty]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- def onTrigger(context, session):
- script_engine = context.getProperty('Script Engine')
- log.info('got Script Engine property: %s' % script_engine)
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- REQUIRE(LogTestController::getInstance().contains("[info] got Script Engine property: python"));
-
- logTestController.reset();
-}
-
-TEST_CASE("Lua: Test Log", "[executescriptLuaLog]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptEngine.getName(), "lua");
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- function onTrigger(context, session)
- log:info('hello from lua')
- end
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- REQUIRE(LogTestController::getInstance().contains(
- "[org::apache::nifi::minifi::processors::ExecuteScript] [info] hello from lua"));
-
- logTestController.reset();
-}
-
-TEST_CASE("Lua: Test Read File", "[executescriptLuaRead]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto putFile = plan->addProcessor("PutFile", "putFile", core::Relationship("success", "description"), true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptEngine.getName(), "lua");
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- read_callback = {}
-
- function read_callback.process(self, input_stream)
- content = input_stream:read()
- log:info('file content: ' .. content)
- return #content
- end
-
- function onTrigger(context, session)
- flow_file = session:get()
-
- if flow_file ~= nil then
- log:info('got flow file: ' .. flow_file:getAttribute('filename'))
- session:read(flow_file, read_callback)
- session:transfer(flow_file, REL_SUCCESS)
- end
- end
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- char putFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *putFileDir = testController.createTempDirectory(putFileDirFmt);
- plan->setProperty(putFile, processors::PutFile::Directory.getName(), putFileDir);
-
- testController.runSession(plan, false);
-
- std::set<provenance::ProvenanceEventRecord *> records = plan->getProvenanceRecords();
- std::shared_ptr<core::FlowFile> record = plan->getCurrentFlowFile();
- REQUIRE(record == nullptr);
- REQUIRE(records.empty());
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- records = plan->getProvenanceRecords();
- record = plan->getCurrentFlowFile();
- testController.runSession(plan, false);
-
- unlink(ss.str().c_str());
-
- REQUIRE(logTestController.contains("[info] file content: tempFile"));
-
- // Verify that file content was preserved
- REQUIRE(!std::ifstream(ss.str()).good());
- std::stringstream movedFile;
- movedFile << putFileDir << "/" << "tstFile.ext";
- REQUIRE(std::ifstream(movedFile.str()).good());
-
- file.open(movedFile.str(), std::ios::in);
- std::string contents((std::istreambuf_iterator<char>(file)),
- std::istreambuf_iterator<char>());
- REQUIRE("tempFile" == contents);
- file.close();
- logTestController.reset();
-}
-
-TEST_CASE("Lua: Test Write File", "[executescriptLuaWrite]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto putFile = plan->addProcessor("PutFile", "putFile", core::Relationship("success", "description"), true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptEngine.getName(), "lua");
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- write_callback = {}
-
- function write_callback.process(self, output_stream)
- new_content = 'hello 2'
- output_stream:write(new_content)
- return #new_content
- end
-
- function onTrigger(context, session)
- flow_file = session:get()
-
- if flow_file ~= nil then
- log:info('got flow file: ' .. flow_file:getAttribute('filename'))
- session:write(flow_file, write_callback)
- session:transfer(flow_file, REL_SUCCESS)
- end
- end
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- char putFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *putFileDir = testController.createTempDirectory(putFileDirFmt);
- plan->setProperty(putFile, processors::PutFile::Directory.getName(), putFileDir);
-
- testController.runSession(plan, false);
-
- std::set<provenance::ProvenanceEventRecord *> records = plan->getProvenanceRecords();
- std::shared_ptr<core::FlowFile> record = plan->getCurrentFlowFile();
- REQUIRE(record == nullptr);
- REQUIRE(records.empty());
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- records = plan->getProvenanceRecords();
- record = plan->getCurrentFlowFile();
- testController.runSession(plan, false);
-
- unlink(ss.str().c_str());
-
- // Verify new content was written
- REQUIRE(!std::ifstream(ss.str()).good());
- std::stringstream movedFile;
- movedFile << putFileDir << "/" << "tstFile.ext";
- REQUIRE(std::ifstream(movedFile.str()).good());
-
- file.open(movedFile.str(), std::ios::in);
- std::string contents((std::istreambuf_iterator<char>(file)),
- std::istreambuf_iterator<char>());
- REQUIRE("hello 2" == contents);
- file.close();
- logTestController.reset();
-}
-
-TEST_CASE("Lua: Test Update Attribute", "[executescriptLuaUpdateAttribute]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::LogAttribute>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto getFile = plan->addProcessor("GetFile", "getFile");
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript",
- core::Relationship("success", "description"),
- true);
- auto logAttribute = plan->addProcessor("LogAttribute", "logAttribute",
- core::Relationship("success", "description"),
- true);
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptEngine.getName(), "lua");
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- function onTrigger(context, session)
- flow_file = session:get()
-
- if flow_file ~= nil then
- log:info('got flow file: ' .. flow_file:getAttribute('filename'))
- flow_file:addAttribute('test_attr', '1')
- attr = flow_file:getAttribute('test_attr')
- log:info('got flow file attr \'test_attr\': ' .. attr)
- flow_file:updateAttribute('test_attr', attr + 1)
- session:transfer(flow_file, REL_SUCCESS)
- end
- end
- )");
-
- char getFileDirFmt[] = "/tmp/ft.XXXXXX";
- char *getFileDir = testController.createTempDirectory(getFileDirFmt);
- plan->setProperty(getFile, processors::GetFile::Directory.getName(), getFileDir);
-
- std::fstream file;
- std::stringstream ss;
- ss << getFileDir << "/" << "tstFile.ext";
- file.open(ss.str(), std::ios::out);
- file << "tempFile";
- file.close();
- plan->reset();
-
- testController.runSession(plan, false);
- testController.runSession(plan, false);
- testController.runSession(plan, false);
-
- REQUIRE(LogTestController::getInstance().contains("key:test_attr value:2"));
-
- logTestController.reset();
-}
-
-TEST_CASE("Lua: Test Create", "[executescriptLuaCreate]") { // NOLINT
- TestController testController;
-
- LogTestController &logTestController = LogTestController::getInstance();
- logTestController.setDebug<TestPlan>();
- logTestController.setDebug<minifi::processors::ExecuteScript>();
-
- auto plan = testController.createPlan();
-
- auto executeScript = plan->addProcessor("ExecuteScript",
- "executeScript");
-
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptEngine.getName(), "lua");
- plan->setProperty(executeScript, processors::ExecuteScript::ScriptBody.getName(), R"(
- function onTrigger(context, session)
- flow_file = session:create(nil)
-
- if flow_file ~= nil then
- log:info('created flow file: ' .. flow_file:getAttribute('filename'))
- session:transfer(flow_file, REL_SUCCESS)
- end
- end
- )");
-
- plan->reset();
-
- testController.runSession(plan, false);
-
- REQUIRE(LogTestController::getInstance().contains("[info] created flow file:"));
-
- logTestController.reset();
-}