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 <[email protected]> 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 <[email protected]> --- 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
