This is an automated email from the ASF dual-hosted git repository. martinzink pushed a commit to branch api_1_1_enrichment in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 1b89551e16c1ba0bb76139a6efc485d5f0cce646 Author: Martin Zink <[email protected]> AuthorDate: Mon Aug 24 16:01:56 2026 +0200 MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes processors MINIFICPP-2889 minifi_process_session_release/minifi_process_session_add --- .../ubuntu_22_04_clang_arm_manifest.json | 166 ++++++++++ CONTROLLERS.md | 1 - PROCESSORS.md | 56 ++++ cmake/Fetchstduuid.cmake | 34 +++ cmake/MiNiFiOptions.cmake | 1 + .../cpp-extension-lib/include/api/core/FlowFile.h | 6 + .../include/api/core/ProcessSession.h | 6 + .../mocklib/include/MockProcessSession.h | 7 + .../mocklib/src/MockProcessSession.cpp | 14 + .../mocklib/src/mock-minifi-api.cpp | 13 + .../cpp-extension-lib/src/core/ProcessSession.cpp | 20 ++ libminifi/include/core/ProcessSession.h | 3 + libminifi/include/core/Processor.h | 8 + libminifi/include/utils/minifi-api-utils.h | 3 + libminifi/src/core/ProcessSession.cpp | 34 ++- libminifi/src/core/Processor.cpp | 17 ++ libminifi/src/core/extension/ApiVersion.cpp | 2 +- libminifi/src/minifi-api.cpp | 39 +++ libminifi/test/unit/ProcessSessionTests.cpp | 32 ++ minifi-api/include/minifi-api.h | 25 +- minifi-api/include/minifi-cpp/core/FlowFile.h | 7 + .../include/minifi-cpp/core/ProcessSession.h | 3 + minifi-api/minifi-api.def | 3 + .../minifi_enrichment/.cargo/config.toml | 5 + .../extensions/minifi_enrichment/.gitignore | 7 + .../extensions/minifi_enrichment/Cargo.toml | 21 ++ .../minifi_enrichment/features/enrichment.feature | 47 +++ .../features/enrichment_restart.feature | 71 +++++ .../minifi_enrichment/features/environment.py | 40 +++ .../minifi_enrichment/features/steps/steps.py | 21 ++ .../minifi_enrichment/src/lib.rs} | 32 +- .../minifi_enrichment/src/processors/attributes.rs | 30 ++ .../src/processors/fork_enrichment.rs | 132 ++++++++ .../fork_enrichment/fork_enrichment_def.rs | 27 ++ .../src/processors/join_enrichment_attributes.rs | 339 +++++++++++++++++++++ .../join_enrichment_attributes_def.rs | 48 +++ .../minifi_enrichment/src/processors/mod.rs} | 25 +- minifi_rust/minifi_native/src/api.rs | 2 + .../minifi_native/src/api/flow_file_store.rs | 194 ++++++++++++ .../minifi_native/src/api/process_session.rs | 24 ++ .../minifi_native/src/c_ffi/c_ffi_flow_file.rs | 32 +- .../src/c_ffi/c_ffi_process_session.rs | 50 ++- minifi_rust/minifi_native/src/lib.rs | 6 +- .../minifi_native/src/mock/mock_flow_file.rs | 10 + .../minifi_native/src/mock/mock_process_session.rs | 14 + 45 files changed, 1622 insertions(+), 55 deletions(-) diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json index 5f72378a9..d4139e9fb 100644 --- a/.github/references/ubuntu_22_04_clang_arm_manifest.json +++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json @@ -4144,6 +4144,89 @@ "version": "1.0.0" } }, +{ + "bundles": { + "componentManifest": { + "processors": [ + { + "propertyDescriptors": { + "Max Batch Size": { + "name": "Max Batch Size", + "description": "The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "false", + "supportedRelationships": [ + { + "name": "enrichment", + "description": "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + }, + { + "name": "original", + "description": "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + } + ], + "typeDescription": "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' [...] + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.enrichment.ForkEnrichment" + }, + { + "propertyDescriptors": { + "Max Batch Size": { + "name": "Max Batch Size", + "description": "The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Timeout": { + "name": "Timeout", + "description": "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship.", + "validator": "TIME_PERIOD_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "true", + "supportedRelationships": [ + { + "name": "invalid", + "description": "Any FlowFiles without the requisite attributes will be routed here" + }, + { + "name": "joined", + "description": "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship" + }, + { + "name": "original", + "description": "Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles." + }, + { + "name": "timeout", + "description": "If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship." + } + ], + "typeDescription": "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's).", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.enrichment.JoinEnrichmentAttributes" + } + ] + }, + "group": "org.apache.nifi.minifi", + "artifact": "minifi-enrichment", + "version": "1.0.0" + } +}, { "bundles": { "componentManifest": { @@ -12104,6 +12187,89 @@ "version": "1.0.0" } }, +{ + "bundles": { + "componentManifest": { + "processors": [ + { + "propertyDescriptors": { + "Batch Size": { + "name": "Batch Size", + "description": "The maximum number of FlowFiles to fork in each trigger", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "false", + "supportedRelationships": [ + { + "name": "enrichment", + "description": "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + }, + { + "name": "original", + "description": "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + } + ], + "typeDescription": "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' [...] + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "minifi_enrichment.processors.fork_enrichment.ForkEnrichment" + }, + { + "propertyDescriptors": { + "Batch Size": { + "name": "Batch Size", + "description": "The maximum number of FlowFiles to process in each trigger", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Timeout": { + "name": "Timeout", + "description": "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship.", + "validator": "TIME_PERIOD_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "true", + "supportedRelationships": [ + { + "name": "invalid", + "description": "Any FlowFiles without the requisite attributes will be routed here" + }, + { + "name": "joined", + "description": "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship" + }, + { + "name": "original", + "description": "Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles." + }, + { + "name": "timeout", + "description": "If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship." + } + ], + "typeDescription": "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's).", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "minifi_enrichment.processors.join_enrichment_attributes.JoinEnrichmentAttributes" + } + ] + }, + "group": "org.apache.nifi.minifi.rust.enrichment", + "artifact": "minifi_enrichment", + "version": "0.1.0" + } +}, { "bundles": { "componentManifest": { diff --git a/CONTROLLERS.md b/CONTROLLERS.md index 28fce43e1..c3ac2fccf 100644 --- a/CONTROLLERS.md +++ b/CONTROLLERS.md @@ -393,4 +393,3 @@ In the list below, the names of required properties appear in bold. Any other pr | **Pretty Print XML** | false | true<br/>false | Specifies whether or not the XML should be pretty printed [...] | **Name of Record Tag** | | | Specifies the name of the XML record tag wrapping the record fields. [...] | **Name of Root Tag** | | | Specifies the name of the XML root tag wrapping the record set. [...] - diff --git a/PROCESSORS.md b/PROCESSORS.md index 4c7705b2a..68487069c 100644 --- a/PROCESSORS.md +++ b/PROCESSORS.md @@ -46,12 +46,14 @@ limitations under the License. - [FetchSFTP](#FetchSFTP) - [FetchSmb](#FetchSmb) - [FocusArchiveEntry](#FocusArchiveEntry) +- [ForkEnrichment](#ForkEnrichment) - [GenerateFlowFile](#GenerateFlowFile) - [GetCouchbaseKey](#GetCouchbaseKey) - [GetFile](#GetFile) - [GetTCP](#GetTCP) - [HashContent](#HashContent) - [InvokeHTTP](#InvokeHTTP) +- [JoinEnrichmentAttributes](#JoinEnrichmentAttributes) - [JoltTransformJSON](#JoltTransformJSON) - [ListAzureBlobStorage](#ListAzureBlobStorage) - [ListAzureDataLakeStorage](#ListAzureDataLakeStorage) @@ -1067,6 +1069,35 @@ In the list below, the names of required properties appear in bold. Any other pr | success | success operational on the flow record | +## ForkEnrichment + +### Description + +Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' relationship. + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|----------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------| +| Max Batch Size | | | The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once. | + +### Relationships + +| Name | Description | +|------------|------------------------------------------------------------------------------------------------------------| +| enrichment | A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes. | +| original | The incoming FlowFile will be routed to this relationship, after adding appropriate attributes. | + +### Output Attributes + +| Attribute | Relationship | Description | +|---------------------|----------------------|---------------------------------------------------------------------------------------------------| +| enrichment.role | enrichment, original | The role to use for enrichment. This will either be ORIGINAL or ENRICHMENT. | +| enrichment.group.id | enrichment, original | The Group ID to use in order to correlate the 'original' FlowFile with the 'enrichment' FlowFile. | + + ## GenerateFlowFile ### Description @@ -1277,6 +1308,31 @@ In the list below, the names of required properties appear in bold. Any other pr | invokehttp.tx.id | success, response, retry, no retry | The transaction ID that is returned after reading the response | +## JoinEnrichmentAttributes + +### Description + +Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's). + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|----------------|---------------|------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Timeout | | | Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship. | +| Max Batch Size | | | The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once. | + +### Relationships + +| Name | Description | +|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| invalid | Any FlowFiles without the requisite attributes will be routed here | +| joined | The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship | +| original | Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles. | +| timeout | If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship. | + + ## JoltTransformJSON ### Description diff --git a/cmake/Fetchstduuid.cmake b/cmake/Fetchstduuid.cmake new file mode 100644 index 000000000..7c13d8e43 --- /dev/null +++ b/cmake/Fetchstduuid.cmake @@ -0,0 +1,34 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +if (NOT stduuid_FOUND) + set(UUID_SYSTEM_GENERATOR "ON") + set(stduuid_FOUND "YES" CACHE STRING "" FORCE) + set(stduuid_INCLUDE_DIR "${CMAKE_BINARY_DIR}/_deps/stduuid/" CACHE STRING "" FORCE) + if(NOT EXISTS "${stduuid_INCLUDE_DIR}/stduuid/uuid.hpp") + file(DOWNLOAD "https://github.com/mariusbancila/stduuid/releases/download/v1.2.3/uuid.h"; "${stduuid_INCLUDE_DIR}/stduuid/uuid.hpp" + EXPECTED_HASH SHA256=8b329afa7e099e632c2e992e02ddb9fc4627c772dfd5fd42b069752ea0f8ec7f) + endif() +endif() + +if(NOT TARGET stduuid::stduuid) + add_library(stduuid::stduuid INTERFACE IMPORTED) + set_target_properties(stduuid::stduuid PROPERTIES + INTERFACE_INCLUDE_DIRECTORIES "${stduuid_INCLUDE_DIR}") +endif() diff --git a/cmake/MiNiFiOptions.cmake b/cmake/MiNiFiOptions.cmake index 7a898a765..a385c1623 100644 --- a/cmake/MiNiFiOptions.cmake +++ b/cmake/MiNiFiOptions.cmake @@ -120,6 +120,7 @@ add_minifi_option(ENABLE_CONTROLLER "Enables the build of MiNiFi controller bina add_minifi_option(ENABLE_LLAMACPP "Enables llama.cpp support." ON) add_minifi_option(ENABLE_OPC "Instructs the build system to enable the OPC extension" ON) add_minifi_option(MINIFI_RUST "Enables the build of rust based extensions." OFF) +add_minifi_option(MINIFI_EXTENSION_ENRICHMENT "Enables the build of enrichment extension" ON) add_minifi_option(MINIFI_LMDB "Enables the LMDB extension." OFF) set_minifi_cache_variable(CUSTOM_MALLOC OFF "Overwrite malloc implementation.") diff --git a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h index c2a5fb424..0c953f80c 100644 --- a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h +++ b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h @@ -39,4 +39,10 @@ struct EnsureMovedFromDeleter { using FlowFile = std::unique_ptr<minifi_flow_file, EnsureMovedFromDeleter>; +struct NoopStashedFlowFileDeleter { + void operator()(minifi_stashed_flow_file*) const noexcept {} +}; + +using StashedFlowFile = std::unique_ptr<minifi_stashed_flow_file, NoopStashedFlowFileDeleter>; + } // namespace org::apache::nifi::minifi::api::core diff --git a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h index 7212046db..f8bd90081 100644 --- a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h +++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h @@ -38,12 +38,15 @@ class ProcessSession { ProcessSession& operator=(const ProcessSession&) = delete; ProcessSession& operator=(ProcessSession&&) = delete; + virtual FlowFile clone(const FlowFile& flow_file) = 0; virtual FlowFile create(const FlowFile* parent = nullptr) = 0; virtual FlowFile get() = 0; virtual void penalize(FlowFile& ff) = 0; virtual void transfer(FlowFile ff, const minifi::core::Relationship& relationship) = 0; virtual void remove(FlowFile ff) = 0; + virtual StashedFlowFile stash(FlowFile ff) = 0; + virtual FlowFile unstash(StashedFlowFile stashed_flow_file) = 0; virtual void write(FlowFile& flow, const io::OutputStreamCallback& callback) = 0; virtual void read(FlowFile& flow, const io::InputStreamCallback& callback) = 0; @@ -64,10 +67,13 @@ class CffiProcessSession : public ProcessSession { explicit CffiProcessSession(minifi_process_session* impl): impl_(impl) {} FlowFile create(const FlowFile* parent = nullptr) override; + FlowFile clone(const FlowFile& flow_file) override; FlowFile get() override; void penalize(FlowFile& ff) override; void transfer(FlowFile ff, const minifi::core::Relationship& relationship) override; void remove(FlowFile ff) override; + StashedFlowFile stash(FlowFile ff) override; + FlowFile unstash(StashedFlowFile stashed_flow_file) override; void write(FlowFile& flow, const io::OutputStreamCallback& callback) override; void read(FlowFile& flow, const io::InputStreamCallback& callback) override; diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h index ad3aeb109..e06f9f539 100644 --- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h +++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h @@ -45,6 +45,10 @@ struct minifi_flow_file { std::string id; }; +struct minifi_stashed_flow_file { + std::unique_ptr<minifi_flow_file> flow_file; +}; + namespace org::apache::nifi::minifi::mock { class MockProcessSession : public api::core::ProcessSession { @@ -57,9 +61,12 @@ class MockProcessSession : public api::core::ProcessSession { api::core::FlowFile create(const api::core::FlowFile* parent) override; api::core::FlowFile get() override; + api::core::FlowFile clone(const api::core::FlowFile& flow_file) override; void penalize(api::core::FlowFile& ff) override; void transfer(api::core::FlowFile ff, const minifi::core::Relationship& relationship) override; void remove(api::core::FlowFile ff) override; + api::core::StashedFlowFile stash(api::core::FlowFile ff) override; + api::core::FlowFile unstash(api::core::StashedFlowFile stashed_flow_file) override; void write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) override; void read(api::core::FlowFile& ff, const io::InputStreamCallback& callback) override; void setAttribute(api::core::FlowFile& ff, std::string_view key, std::string value) override; diff --git a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp index 6af5178e5..30b53046f 100644 --- a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp +++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp @@ -33,6 +33,11 @@ api::core::FlowFile MockProcessSession::get() { return api::core::FlowFile{ff.release()}; } +api::core::FlowFile MockProcessSession::clone(const api::core::FlowFile& flow_file) { + return api::core::FlowFile{new minifi_flow_file(*flow_file.get())}; +} + + void MockProcessSession::penalize(api::core::FlowFile& ff) { ff->is_penalized = true; } @@ -45,6 +50,15 @@ void MockProcessSession::remove(api::core::FlowFile ff) { removed_flow_files_.push_back(std::unique_ptr<minifi_flow_file>(ff.release())); } +api::core::StashedFlowFile MockProcessSession::stash(api::core::FlowFile ff) { + return api::core::StashedFlowFile{new minifi_stashed_flow_file{std::unique_ptr<minifi_flow_file>(ff.release())}}; +} + +api::core::FlowFile MockProcessSession::unstash(api::core::StashedFlowFile stashed_flow_file) { + const std::unique_ptr<minifi_stashed_flow_file> owned{stashed_flow_file.release()}; + return api::core::FlowFile{owned->flow_file.release()}; +} + void MockProcessSession::write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) { const auto stream = std::make_shared<MockOutputStream>(ff->content); callback(stream); diff --git a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp index 5647bfa11..e244ba51c 100644 --- a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp +++ b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp @@ -17,6 +17,7 @@ #include <stdexcept> +#include "MockProcessSession.h" #include "minifi-api.h" extern "C" { @@ -75,6 +76,14 @@ minifi_status minifi_process_session_remove(minifi_process_session*, MINIFI_OWNE throw std::runtime_error("Not implemented"); } +minifi_status minifi_process_session_stash(minifi_process_session*, MINIFI_OWNED minifi_flow_file*, minifi_stashed_flow_file**) { + throw std::runtime_error("Not implemented"); +} + +minifi_status minifi_process_session_unstash(minifi_process_session*, MINIFI_OWNED minifi_stashed_flow_file*, minifi_flow_file**) { + throw std::runtime_error("Not implemented"); +} + minifi_status minifi_process_session_read(minifi_process_session*, minifi_flow_file*, int64_t (*)(void* user_ctx, minifi_input_stream*), void*) { throw std::runtime_error("Not implemented"); } @@ -146,4 +155,8 @@ minifi_status minifi_process_context_report_metrics(minifi_process_context*, siz throw std::runtime_error("Not implemented"); } +minifi_flow_file* minifi_process_session_clone(minifi_process_session*, minifi_flow_file*) { + throw std::runtime_error("Not implemented"); +} + } // extern "C" diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp index 41b8be081..0f232e5c8 100644 --- a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp +++ b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp @@ -77,6 +77,10 @@ FlowFile CffiProcessSession::create(const FlowFile* parent) { return FlowFile{minifi_process_session_create(impl_, parent ? parent->get() : nullptr)}; } +FlowFile CffiProcessSession::clone(const FlowFile& flow_file) { + return FlowFile{minifi_process_session_clone(impl_, flow_file.get())}; +} + void CffiProcessSession::penalize(FlowFile& ff) { if (MINIFI_STATUS_SUCCESS != minifi_process_session_penalize(impl_, ff.get())) { throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to penalize flowfile"); @@ -96,6 +100,22 @@ void CffiProcessSession::remove(FlowFile ff) { } } +StashedFlowFile CffiProcessSession::stash(FlowFile ff) { + minifi_stashed_flow_file* stashed = nullptr; + if (MINIFI_STATUS_SUCCESS != minifi_process_session_stash(impl_, ff.release(), &stashed)) { + throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to stash flowfile"); + } + return StashedFlowFile{stashed}; +} + +FlowFile CffiProcessSession::unstash(StashedFlowFile stashed_flow_file) { + minifi_flow_file* ff = nullptr; + if (MINIFI_STATUS_SUCCESS != minifi_process_session_unstash(impl_, stashed_flow_file.release(), &ff)) { + throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to unstash flowfile"); + } + return FlowFile{ff}; +} + void CffiProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback& callback) { const auto status = minifi_process_session_write( impl_, diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h index 72494dd1b..f88402d8d 100644 --- a/libminifi/include/core/ProcessSession.h +++ b/libminifi/include/core/ProcessSession.h @@ -79,6 +79,9 @@ class ProcessSessionImpl : public ReferenceContainerImpl, public virtual Process void remove(const std::shared_ptr<core::FlowFile> &flow) override; + StashedFlowFile stash(std::shared_ptr<core::FlowFile> flow_file) override; + std::shared_ptr<core::FlowFile> unstash(StashedFlowFile stashed_flow_file) override; + std::shared_ptr<io::InputStream> getFlowFileContentStream(const core::FlowFile& flow_file) override; int64_t read(const std::shared_ptr<core::FlowFile>& flow_file, const io::InputStreamCallback& callback) override; diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h index 4f48c0938..9e47a6118 100644 --- a/libminifi/include/core/Processor.h +++ b/libminifi/include/core/Processor.h @@ -123,6 +123,11 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ void setLoggerCallback(const std::function<void(logging::LOG_LEVEL level, const std::string& message)>& callback); void restore(const std::shared_ptr<FlowFile>& file) override; + // Stashing keeps a flow file owned by this processor across onTrigger invocations (used by the + // stable C API's stash/unstash). The owning shared_ptr lives here; sessions hand it in and out. + void stashFlowFile(std::shared_ptr<FlowFile> flow_file); + std::shared_ptr<FlowFile> unstashFlowFile(const FlowFile* flow_file); + static constexpr auto DynamicProperties = std::array<DynamicPropertyDefinition, 0>{}; static constexpr auto OutputAttributes = std::array<OutputAttributeReference, 0>{}; @@ -159,6 +164,9 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ mutable std::mutex mutex_; std::atomic<std::chrono::steady_clock::time_point> yield_expiration_{}; + // Flow files stashed with this processor (owning), keyed by identity via unstashFlowFile. + std::vector<std::shared_ptr<FlowFile>> stashed_flow_files_; + // must hold the graphMutex void updateReachability(const std::lock_guard<std::mutex>& graph_lock, bool force = false); diff --git a/libminifi/include/utils/minifi-api-utils.h b/libminifi/include/utils/minifi-api-utils.h index f98b60813..54b77bfd7 100644 --- a/libminifi/include/utils/minifi-api-utils.h +++ b/libminifi/include/utils/minifi-api-utils.h @@ -73,6 +73,9 @@ MINIFI_API_MAP(minifi_extension, minifi::core::extension::Extension); MINIFI_API_MAP(minifi_extension_context, minifi::core::extension::Extension::Context); MINIFI_API_MAP(minifi_logger, std::shared_ptr<minifi::core::logging::Logger>); MINIFI_API_MAP(minifi_flow_file, std::shared_ptr<minifi::core::FlowFile>); +// A stashed flow file handle is an opaque, non-owning identity token: the raw FlowFile whose owning +// shared_ptr lives in the processor's stash. See core::StashedFlowFile / Processor::stashFlowFile. +MINIFI_API_MAP(minifi_stashed_flow_file, minifi::core::FlowFile); #undef MINIFI_API_MAP diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp index 4352f7b41..c1ca7356e 100644 --- a/libminifi/src/core/ProcessSession.cpp +++ b/libminifi/src/core/ProcessSession.cpp @@ -209,6 +209,35 @@ void ProcessSessionImpl::remove(const std::shared_ptr<core::FlowFile> &flow) { provenance_report_->drop(*flow, reason); } +StashedFlowFile ProcessSessionImpl::stash(std::shared_ptr<core::FlowFile> flow_file) { + logger_->log_trace("Stashing flow file with UUID: {}", flow_file->getUUIDStr()); + const auto uuid = flow_file->getUUID(); + // The session forgets this flow file: on commit it is neither routed, persisted, nor deleted. + updated_flowfiles_.erase(uuid); + updated_relationships_.erase(uuid); + added_flowfiles_.erase(uuid); + std::erase_if(deleted_flowfiles_, [&uuid](const auto& ff) { return ff->getUUID() == uuid; }); + + core::FlowFile* identity = flow_file.get(); + process_context_->getProcessor().stashFlowFile(std::move(flow_file)); + return StashedFlowFile{identity}; +} + +std::shared_ptr<core::FlowFile> ProcessSessionImpl::unstash(StashedFlowFile stashed_flow_file) { + auto record = process_context_->getProcessor().unstashFlowFile(stashed_flow_file.flow_file); + if (!record) { + return nullptr; + } + logger_->log_trace("Unstashing flow file with UUID: {}", record->getUUIDStr()); + const utils::Identifier uuid = record->getUUID(); + if (updated_flowfiles_.contains(uuid)) { + throw Exception(ExceptionType::PROCESSOR_EXCEPTION, "Mustn't unstash a file that was provided by this session"); + } + added_flowfiles_[uuid].flow_file = record; + record->setDeleted(false); + return record; +} + void ProcessSessionImpl::putAttribute(core::FlowFile& flow_file, std::string_view key, const std::string& value) { flow_file.setAttribute(key, value); std::string details = fmt::format("{} modify flow record {} attribute {}:{}", process_context_->getProcessor().getName(), flow_file.getUUIDStr(), key, value); @@ -233,8 +262,11 @@ void ProcessSessionImpl::transfer(const std::shared_ptr<core::FlowFile>& flow, c utils::Identifier uuid = flow->getUUID(); if (auto it = added_flowfiles_.find(uuid); it != added_flowfiles_.end()) { it->second.rel = &*relationships_.insert(relationship).first; - } else { + } else if (updated_flowfiles_.contains(uuid)) { updated_relationships_[uuid] = &*relationships_.insert(relationship).first; + } else { + throw Exception(PROCESS_SESSION_EXCEPTION, + "Cannot transfer flow file " + flow->getUUIDStr() + " that was not obtained from or added to this session"); } flow->setDeleted(false); } diff --git a/libminifi/src/core/Processor.cpp b/libminifi/src/core/Processor.cpp index b20004a09..59675699f 100644 --- a/libminifi/src/core/Processor.cpp +++ b/libminifi/src/core/Processor.cpp @@ -543,6 +543,23 @@ void Processor::restore(const std::shared_ptr<FlowFile>& file) { impl_->restore(file); } +void Processor::stashFlowFile(std::shared_ptr<FlowFile> flow_file) { + const std::lock_guard<std::mutex> lock(mutex_); + stashed_flow_files_.push_back(std::move(flow_file)); +} + +std::shared_ptr<FlowFile> Processor::unstashFlowFile(const FlowFile* flow_file) { + const std::lock_guard<std::mutex> lock(mutex_); + for (auto it = stashed_flow_files_.begin(); it != stashed_flow_files_.end(); ++it) { + if (it->get() == flow_file) { + auto result = std::move(*it); + stashed_flow_files_.erase(it); + return result; + } + } + return nullptr; +} + const std::unordered_map<Connection*, std::unordered_set<Processor*>>& Processor::reachable_processors() const { return reachable_processors_; } diff --git a/libminifi/src/core/extension/ApiVersion.cpp b/libminifi/src/core/extension/ApiVersion.cpp index e7930fc11..d9d6a91e2 100644 --- a/libminifi/src/core/extension/ApiVersion.cpp +++ b/libminifi/src/core/extension/ApiVersion.cpp @@ -22,7 +22,7 @@ namespace org::apache::nifi::minifi::core::extension { static uint32_t agent_api_version{MINIFI_API_VERSION}; -static uint32_t min_supported_api_version{MINIFI_API_VERSION}; +static uint32_t min_supported_api_version = 2; // Since version 2 everything was additive only uint32_t getAgentApiVersion() { return agent_api_version; diff --git a/libminifi/src/minifi-api.cpp b/libminifi/src/minifi-api.cpp index 0fae27f2b..c7a3b6eee 100644 --- a/libminifi/src/minifi-api.cpp +++ b/libminifi/src/minifi-api.cpp @@ -442,6 +442,15 @@ MINIFI_OWNED minifi_flow_file* minifi_process_session_create(minifi_process_sess return nullptr; } +MINIFI_OWNED minifi_flow_file* minifi_process_session_clone(minifi_process_session* session, minifi_flow_file* flow_file) { + gsl_Assert(session); + gsl_Assert(flow_file); + if (const auto ff = toCpp(session)->clone(*toRawFlowFile(flow_file))) { + return toC(new std::shared_ptr<minifi::core::FlowFile>(ff)); // NOLINT(cppcoreguidelines-owning-memory) + } + return nullptr; +} + minifi_status minifi_process_session_penalize(minifi_process_session* session, minifi_flow_file* flowfile) { gsl_Assert(session); gsl_Assert(flowfile); @@ -479,6 +488,36 @@ minifi_status minifi_process_session_remove(minifi_process_session* session, MIN } } +enum minifi_status minifi_process_session_stash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flow_file, struct minifi_stashed_flow_file** stashed_flow_file) { + gsl_Assert(session); + gsl_Assert(flow_file); + gsl_Assert(stashed_flow_file); + const std::unique_ptr<std::shared_ptr<minifi::core::FlowFile>> owned{toCpp(flow_file)}; + try { + *stashed_flow_file = toC(toCpp(session)->stash(std::move(*owned)).flow_file); + return MINIFI_STATUS_SUCCESS; + } catch (...) { + return MINIFI_STATUS_UNKNOWN_ERROR; + } +} + +minifi_status minifi_process_session_unstash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_stashed_flow_file* stashed_flow_file, struct minifi_flow_file** flow_file) { + gsl_Assert(session); + gsl_Assert(flow_file); + gsl_Assert(stashed_flow_file); + try { + auto unstashed = toCpp(session)->unstash(minifi::core::StashedFlowFile{toCpp(stashed_flow_file)}); + if (!unstashed) { + return MINIFI_STATUS_UNKNOWN_ERROR; + } + *flow_file = toC(new std::shared_ptr<minifi::core::FlowFile>(std::move(unstashed))); // NOLINT(cppcoreguidelines-owning-memory) + return MINIFI_STATUS_SUCCESS; + } + catch (...) { + return MINIFI_STATUS_UNKNOWN_ERROR; + } +} + minifi_status minifi_process_session_read(minifi_process_session* session, minifi_flow_file* flowfile, int64_t(*cb)(void* user_ctx, minifi_input_stream*), void* user_ctx) { gsl_Assert(session); gsl_Assert(flowfile); diff --git a/libminifi/test/unit/ProcessSessionTests.cpp b/libminifi/test/unit/ProcessSessionTests.cpp index 8c7e99bd6..0bbfc4f0d 100644 --- a/libminifi/test/unit/ProcessSessionTests.cpp +++ b/libminifi/test/unit/ProcessSessionTests.cpp @@ -166,3 +166,35 @@ TEST_CASE("Stable C API does not leak the flow file handle", "[minifi-api][flowf // FlowFile alive and this assertion fails. REQUIRE(weak_flow_file.expired()); } + +TEST_CASE("Stable C API: stash keeps a flow file across a commit and unstash brings it back", "[minifi-api][flowfilehandle]") { + Fixture fixture; + auto& process_session = fixture.processSession(); + auto* c_session = minifi::utils::toC(&process_session); + + auto created = process_session.create(); + process_session.transfer(created, Success); + process_session.commit(); + + std::weak_ptr<minifi::core::FlowFile> weak_flow_file = created; + created.reset(); + + auto* handle = minifi_process_session_get(c_session); + REQUIRE(handle != nullptr); + + // Stash detaches the flow file from the session; the returned stashed handle keeps it alive + // across the commit (it is neither routed nor deleted). + minifi_stashed_flow_file* stashed = nullptr; + REQUIRE(minifi_process_session_stash(c_session, handle, &stashed) == MINIFI_STATUS_SUCCESS); + REQUIRE(stashed != nullptr); + REQUIRE_NOTHROW(process_session.commit()); + REQUIRE_FALSE(weak_flow_file.expired()); + + // Unstash hands it back as a MINIFI_OWNED handle; removing it and committing frees it for good. + minifi_flow_file* unstashed = nullptr; + REQUIRE(minifi_process_session_unstash(c_session, stashed, &unstashed) == MINIFI_STATUS_SUCCESS); + REQUIRE(unstashed != nullptr); + REQUIRE(minifi_process_session_remove(c_session, unstashed) == MINIFI_STATUS_SUCCESS); + process_session.commit(); + REQUIRE(weak_flow_file.expired()); +} diff --git a/minifi-api/include/minifi-api.h b/minifi-api/include/minifi-api.h index f9d1fe078..177588833 100644 --- a/minifi-api/include/minifi-api.h +++ b/minifi-api/include/minifi-api.h @@ -43,7 +43,7 @@ extern "C" { #define MINIFI_PROXY_CONFIGURATION_SERVICE_INTERFACE_PROPERTY_TYPE "org.apache.nifi.minifi.controllers.ProxyConfigurationServiceInterface" enum : uint32_t { - MINIFI_API_VERSION = 2 + MINIFI_API_VERSION = 3 }; enum minifi_io_status : int64_t { @@ -100,6 +100,8 @@ struct minifi_output_stream; struct minifi_extension; struct minifi_extension_context; +struct minifi_stashed_flow_file; // MINIFI_API_VERSION >= 3 + enum minifi_status : uint32_t { MINIFI_STATUS_SUCCESS = 0, MINIFI_STATUS_UNKNOWN_ERROR = 1, @@ -243,12 +245,33 @@ bool minifi_logger_should_log(struct minifi_logger*, enum minifi_log_level); MINIFI_OWNED struct minifi_flow_file* minifi_process_session_get(struct minifi_process_session*); MINIFI_OWNED struct minifi_flow_file* minifi_process_session_create(struct minifi_process_session* session, MINIFI_NULLABLE struct minifi_flow_file* parent_flowfile); +MINIFI_OWNED struct minifi_flow_file* minifi_process_session_clone(struct minifi_process_session* session, + MINIFI_NULLABLE struct minifi_flow_file* flow_file); enum minifi_status minifi_process_session_penalize(struct minifi_process_session* session, struct minifi_flow_file* flowfile); enum minifi_status minifi_process_session_transfer(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flowfile, struct minifi_string_view relationship_name); enum minifi_status minifi_process_session_remove(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flowfile); +// stash/unstash let an extension keep a flow file across onTrigger invocations without the legacy +// self-relationship trick. The flow file's record/content stay in the repositories throughout; what +// moves is where it is owned. +// +// stash: takes a MINIFI_OWNED flow file (e.g. one from get()) out of the current session - on commit +// it is neither routed, persisted, nor deleted - and hands its ownership to the processor, which +// outlives the session. It writes an opaque handle to *stashed_flow_file identifying the stashed +// flow file. The handle is a lightweight token, not a new allocation: there is nothing to free. +// unstash: hands a stashed flow file back to a (later) session. It writes a MINIFI_OWNED flow file to +// *flow_file, which is owned by that session again and must be transferred or removed as usual. +// The stashed_flow_file token is consumed and must not be used again. +// +// Because the processor owns stashed flow files, there is no explicit free: a token that is never +// unstashed simply leaves its flow file in the processor, which reclaims it when the processor is +// torn down (the record/content stay in the repositories and are recovered on next startup). +enum minifi_status minifi_process_session_stash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flow_file, struct minifi_stashed_flow_file** stashed_flow_file); +enum minifi_status minifi_process_session_unstash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_stashed_flow_file* stashed_flow_file, struct minifi_flow_file** flow_file); + + enum minifi_status minifi_process_session_read(struct minifi_process_session*, struct minifi_flow_file*, int64_t (*cb)(void* user_ctx, struct minifi_input_stream*), void* user_ctx); enum minifi_status minifi_process_session_write(struct minifi_process_session*, struct minifi_flow_file*, diff --git a/minifi-api/include/minifi-cpp/core/FlowFile.h b/minifi-api/include/minifi-cpp/core/FlowFile.h index 951829b45..8a026478c 100644 --- a/minifi-api/include/minifi-cpp/core/FlowFile.h +++ b/minifi-api/include/minifi-cpp/core/FlowFile.h @@ -81,4 +81,11 @@ class FlowFile : public virtual CoreComponent, public virtual ReferenceContainer static std::shared_ptr<FlowFile> create(); }; +// A non-owning token identifying a flow file stashed with a processor (see +// ProcessSession::stash / Processor::stashFlowFile). The owning shared_ptr lives in the processor; +// this only records the flow file's identity so unstash() can retrieve it. +struct StashedFlowFile { + core::FlowFile* flow_file; +}; + } // namespace org::apache::nifi::minifi::core diff --git a/minifi-api/include/minifi-cpp/core/ProcessSession.h b/minifi-api/include/minifi-cpp/core/ProcessSession.h index 5b77d8422..3820e7a0f 100644 --- a/minifi-api/include/minifi-cpp/core/ProcessSession.h +++ b/minifi-api/include/minifi-cpp/core/ProcessSession.h @@ -83,6 +83,9 @@ class ProcessSession : public virtual ReferenceContainer { // Penalize the flow virtual void penalize(const std::shared_ptr<core::FlowFile> &flow) = 0; + virtual StashedFlowFile stash(std::shared_ptr<core::FlowFile> flow_file) = 0; + virtual std::shared_ptr<core::FlowFile> unstash(StashedFlowFile stashed_flow_file) = 0; + virtual bool outgoingConnectionsFull(const std::string& relationship) = 0; /** diff --git a/minifi-api/minifi-api.def b/minifi-api/minifi-api.def index a0a0b74ad..2670fe8e7 100644 --- a/minifi-api/minifi-api.def +++ b/minifi-api/minifi-api.def @@ -14,6 +14,7 @@ EXPORTS minifi_process_context_get_ssl_data_from_property minifi_process_context_report_metrics minifi_process_context_set_trigger_when_empty + minifi_process_session_clone minifi_process_session_create minifi_process_session_get minifi_process_session_get_flow_file_attribute @@ -24,7 +25,9 @@ EXPORTS minifi_process_session_read minifi_process_session_remove minifi_process_session_set_flow_file_attribute + minifi_process_session_stash minifi_process_session_transfer + minifi_process_session_unstash minifi_process_session_write minifi_register_controller_service minifi_register_extension diff --git a/minifi_rust/extensions/minifi_enrichment/.cargo/config.toml b/minifi_rust/extensions/minifi_enrichment/.cargo/config.toml new file mode 100644 index 000000000..cb8c02ddc --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/.cargo/config.toml @@ -0,0 +1,5 @@ +[target.aarch64-apple-darwin] +rustflags = ["-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup"] + +[target.x86_64-apple-darwin] +rustflags = ["-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup"] diff --git a/minifi_rust/extensions/minifi_enrichment/.gitignore b/minifi_rust/extensions/minifi_enrichment/.gitignore new file mode 100644 index 000000000..32c9b263b --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/.gitignore @@ -0,0 +1,7 @@ +target +output +features/.venv +features/output +integration-test/features/.venv +integration-test/features/linux_so +integration-test/.venv diff --git a/minifi_rust/extensions/minifi_enrichment/Cargo.toml b/minifi_rust/extensions/minifi_enrichment/Cargo.toml new file mode 100644 index 000000000..71f132fd7 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "minifi_enrichment" +version = "0.1.0" +edition = "2024" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +minifi_native = { path = "../../minifi_native" } +strum = "0.28.0" +walkdir = "2.5.0" +rand = "0.10.0" +hex = "0.4.3" +strum_macros = "0.28.0" +lipsum = "0.9.1" +uuid = { version = "1.24.0", features = ["v4"] } + +[dev-dependencies] +tempfile = "3.22.0" +filetime = "0.2.26" diff --git a/minifi_rust/extensions/minifi_enrichment/features/enrichment.feature b/minifi_rust/extensions/minifi_enrichment/features/enrichment.feature new file mode 100644 index 000000000..1401abc46 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/enrichment.feature @@ -0,0 +1,47 @@ +# 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. + +@MINIFI_EXTENSION_ENRICHMENT @SUPPORTS_WINDOWS +Feature: ForkEnrichment and JoinEnrichmentAttributes + + Scenario: Merges correctly + Given a GenerateFlowFile processor with the "Custom Text" property set to "original_${literal("content")}" + And the scheduling period of the GenerateFlowFile processor is set to "1 hour" + And the "Data Format" property of the GenerateFlowFile processor is set to "Text" + And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false" + + And a ForkEnrichment processor + And a JoinEnrichmentAttributes processor + + And a ReplaceText processor with the "Evaluation Mode" property set to "Entire text" + And the "Replacement Strategy" property of the ReplaceText processor is set to "Always Replace" + And the "Replacement Value" property of the ReplaceText processor is set to "replaced_content" + + And an UpdateAttribute processor with the "extra_prop" property set to "foo" + + And a LogAttribute processor with the "Log Payload" property set to "true" + + And the "success" relationship of the GenerateFlowFile processor is connected to the ForkEnrichment + And the "original" relationship of the ForkEnrichment processor is connected to the JoinEnrichmentAttributes + And the "enrichment" relationship of the ForkEnrichment processor is connected to the ReplaceText + And the "success" relationship of the ReplaceText processor is connected to the UpdateAttribute + And the "success" relationship of the UpdateAttribute processor is connected to the JoinEnrichmentAttributes + And the "joined" relationship of the JoinEnrichmentAttributes processor is connected to the LogAttribute + And JoinEnrichmentAttributes's original relationship is auto-terminated + And LogAttribute's success relationship is auto-terminated + When the MiNiFi instance starts up + Then the Minifi logs contain the following message: "key:enrichment.role value:JOINED" in less than 10 seconds + And the Minifi logs contain the following message: "key:extra_prop value:foo" in less than 1 second + And the Minifi logs contain the following message: "original_content" in less than 1 second diff --git a/minifi_rust/extensions/minifi_enrichment/features/enrichment_restart.feature b/minifi_rust/extensions/minifi_enrichment/features/enrichment_restart.feature new file mode 100644 index 000000000..36d329101 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/enrichment_restart.feature @@ -0,0 +1,71 @@ +# 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. + +@MINIFI_EXTENSION_ENRICHMENT +Feature: JoinEnrichmentAttributes recovers released FlowFiles after an agent restart + + # JoinEnrichmentAttributes releases the first half of a pair and holds it in memory until the + # other half arrives. The released FlowFile is not routed, persisted-as-moved, nor deleted on + # commit, so its record stays in the (default, RocksDB-backed) flow file repository. When the + # agent is restarted the held FlowFile must be recovered from disk and re-enqueued, so that the + # join still completes once its pair finally arrives. + # + # The two halves are produced independently with GetFile + UpdateAttribute (instead of + # ForkEnrichment) so their arrival can be separated across the restart: the ORIGINAL is present + # at startup, and its ENRICHMENT pair is only delivered after the restart. + + Scenario: A released FlowFile held across an agent restart is recovered and joined with its late-arriving pair + Given a GetFile processor with the name "GetOriginal" and the "Input Directory" property set to "/tmp/original_input" + And the scheduling period of the GetOriginal processor is set to "100 ms" + And a UpdateAttribute processor with the name "TagOriginal" and the "enrichment.role" property set to "ORIGINAL" + And the "enrichment.group.id" property of the TagOriginal processor is set to "group-1" + + And a GetFile processor with the name "GetEnrichment" and the "Input Directory" property set to "/tmp/enrichment_input" + And the scheduling period of the GetEnrichment processor is set to "100 ms" + And a UpdateAttribute processor with the name "TagEnrichment" and the "enrichment.role" property set to "ENRICHMENT" + And the "enrichment.group.id" property of the TagEnrichment processor is set to "group-1" + + And a JoinEnrichmentAttributes processor + And a PutFile processor with the "Directory" property set to "/tmp/output" + + And the "success" relationship of the GetOriginal processor is connected to the TagOriginal + And the "success" relationship of the TagOriginal processor is connected to the JoinEnrichmentAttributes + And the "success" relationship of the GetEnrichment processor is connected to the TagEnrichment + And the "success" relationship of the TagEnrichment processor is connected to the JoinEnrichmentAttributes + And the "joined" relationship of the JoinEnrichmentAttributes processor is connected to the PutFile + + And JoinEnrichmentAttributes's original relationship is auto-terminated + And JoinEnrichmentAttributes's invalid relationship is auto-terminated + And JoinEnrichmentAttributes's timeout relationship is auto-terminated + And PutFile's success relationship is auto-terminated + And PutFile's failure relationship is auto-terminated + + # Only the ORIGINAL half exists at startup; the ENRICHMENT half arrives after the restart. + And a directory at "/tmp/original_input" has a file with the content "original_content" + + When the MiNiFi instance starts up + # JoinEnrichmentAttributes gets the ORIGINAL, releases it and holds it waiting for its pair, + # so nothing is joined yet. + Then no files are placed in the "/tmp/output" directory in 5 seconds of running time + + # Graceful stop destroys JoinEnrichmentAttributes while it still holds the released ORIGINAL, + # then restart brings the agent back with the persistent repositories intact. + When MiNiFi is stopped + And MiNiFi is restarted + + # The released ORIGINAL is recovered from the flow file repository and re-enqueued. Once its + # ENRICHMENT pair arrives, the join produces a single FlowFile with the ORIGINAL's content. + And a file with the content "enrichment_content" is placed in "/tmp/enrichment_input" + Then a single file with the content "original_content" is placed in the "/tmp/output" directory in less than 60 seconds diff --git a/minifi_rust/extensions/minifi_enrichment/features/environment.py b/minifi_rust/extensions/minifi_enrichment/features/environment.py new file mode 100644 index 000000000..7e198c4ac --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/environment.py @@ -0,0 +1,40 @@ +# 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 +# +# https://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. + +import os + +from minifi_behave.core.hooks import ( + add_extension_to_minifi_container, + common_after_scenario, + common_before_scenario, +) + + +def before_all(context): + dir_path = os.path.dirname(os.path.realpath(__file__)) + build_path = os.path.normpath(os.path.join(dir_path, "../../../target/release/")) + deps_build_path = os.path.normpath(os.path.join(dir_path, "../../../target/release/deps/")) + add_extension_to_minifi_container("minifi_enrichment", [build_path, deps_build_path], context) + + +def before_scenario(context, scenario): + context.minifi_container_image = "apacheminificpp:minifi_enrichment" + common_before_scenario(context, scenario) + + +def after_scenario(context, scenario): + common_after_scenario(context, scenario) diff --git a/minifi_rust/extensions/minifi_enrichment/features/steps/steps.py b/minifi_rust/extensions/minifi_enrichment/features/steps/steps.py new file mode 100644 index 000000000..ff5817598 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/steps/steps.py @@ -0,0 +1,21 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from minifi_behave.steps import ( + checking_steps, # noqa: F401 + configuration_steps, # noqa: F401 + core_steps, # noqa: F401 + flow_building_steps, # noqa: F401 +) diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs b/minifi_rust/extensions/minifi_enrichment/src/lib.rs similarity index 61% copy from minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs copy to minifi_rust/extensions/minifi_enrichment/src/lib.rs index 580c04626..850e662ac 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs +++ b/minifi_rust/extensions/minifi_enrichment/src/lib.rs @@ -15,25 +15,17 @@ // specific language governing permissions and limitations // under the License. -use crate::api::FlowFile; -use minifi_native_sys::minifi_flow_file; +mod processors; -pub struct CffiFlowFile<'a> { - ptr: *mut minifi_flow_file, - _lifetime: std::marker::PhantomData<&'a ()>, -} +use crate::processors::fork_enrichment::ForkEnrichment; +use crate::processors::join_enrichment_attributes::JoinEnrichmentAttributes; +use minifi_native::{ComplexProcessorType, MultiThreaded, SingleThreaded}; -impl CffiFlowFile<'_> { - pub(crate) fn new(ptr: *mut minifi_flow_file) -> Self { - Self { - ptr, - _lifetime: std::marker::PhantomData, - } - } - - pub(crate) fn get_ptr(&self) -> *mut minifi_flow_file { - self.ptr - } -} - -impl FlowFile for CffiFlowFile<'_> {} +minifi_native::declare_minifi_extension!( + group_name: "org.apache.nifi.minifi.rust.enrichment", + processors: [ + (ComplexProcessorType, MultiThreaded, ForkEnrichment), + (ComplexProcessorType, SingleThreaded, JoinEnrichmentAttributes), + ], + controllers: [] +); diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/attributes.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/attributes.rs new file mode 100644 index 000000000..e9149827d --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/attributes.rs @@ -0,0 +1,30 @@ +use minifi_native::OutputAttribute; +use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames}; + +const ENRICHMENT_ROLE: &str = "enrichment.role"; +const ENRICHMENT_GROUP_ID: &str = "enrichment.group.id"; + +#[derive(Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames, IntoStaticStr)] +#[strum(serialize_all = "UPPERCASE", const_into_str)] +pub(crate) enum Role { + Original, + Enrichment, +} + +pub(crate) const FORK_ROLE_ATTR: OutputAttribute = OutputAttribute { + name: ENRICHMENT_ROLE, + relationships: &["enrichment", "original"], + description: "The role to use for enrichment. This will either be ORIGINAL or ENRICHMENT.", +}; + +pub(crate) const GROUP_ID_ATTR: OutputAttribute = OutputAttribute { + name: ENRICHMENT_GROUP_ID, + relationships: &["enrichment", "original"], + description: "The Group ID to use in order to correlate the 'original' FlowFile with the 'enrichment' FlowFile.", +}; + +pub(crate) const JOIN_ROLE_ATTR: OutputAttribute = OutputAttribute { + name: ENRICHMENT_ROLE, + relationships: &["joined"], + description: "JOINED", +}; diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment.rs new file mode 100644 index 000000000..08caaf7d8 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment.rs @@ -0,0 +1,132 @@ +mod fork_enrichment_def; + +use crate::processors::attributes::{FORK_ROLE_ATTR, GROUP_ID_ATTR}; +use crate::processors::fork_enrichment::fork_enrichment_def::{ENRICHMENT, ORIGINAL}; +use minifi_native::macros::ComponentIdentifier; +use minifi_native::{ + GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule, + Trigger, +}; +use uuid::Uuid; + +#[derive(ComponentIdentifier)] +pub(crate) struct ForkEnrichment { +} + +impl Schedule for ForkEnrichment { + fn schedule<Ctx: GetProperty, L: Logger>( + _context: &Ctx, + _logger: &L, + ) -> Result<Self, MinifiError> + where + Self: Sized, + { + Ok(Self { + }) + } +} + +impl Trigger for ForkEnrichment { + fn trigger<Context, Session, Lggr>( + &self, + _context: &mut Context, + session: &mut Session, + _logger: &Lggr, + ) -> Result<OnTriggerResult, MinifiError> + where + Context: ProcessContext, + Session: ProcessSession<FlowFile = Context::FlowFile>, + Lggr: Logger, + { + let Some(mut original) = session.get() else { + return Ok(OnTriggerResult::Yield); + }; + let mut enrichment = session.clone_ff(&original)?; + session.set_attribute(&mut original, FORK_ROLE_ATTR.name, "ORIGINAL")?; + session.set_attribute(&mut enrichment, FORK_ROLE_ATTR.name, "ENRICHMENT")?; + + let group_id = Uuid::new_v4().to_string(); + session.set_attribute(&mut original, GROUP_ID_ATTR.name, &group_id)?; + session.set_attribute(&mut enrichment, GROUP_ID_ATTR.name, &group_id)?; + + session.transfer(original, ORIGINAL.name)?; + session.transfer(enrichment, ENRICHMENT.name)?; + + Ok(OnTriggerResult::Ok) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use minifi_native::{ + ComponentIdentifier, MockFlowFile, MockLogger, MockProcessContext, MockProcessSession, + }; + + #[test] + fn component_id() { + assert_eq!( + ForkEnrichment::CLASS_NAME, + "minifi_enrichment::processors::fork_enrichment::ForkEnrichment" + ); + assert_eq!(ForkEnrichment::GROUP_NAME, "minifi_enrichment"); + } + + #[test] + fn forks_into_original_and_enrichment_sharing_a_group_id() { + let logger = MockLogger::new(); + let mut context = MockProcessContext::new(); + let processor = ForkEnrichment::schedule(&context, &logger).unwrap(); + + let mut session = MockProcessSession::new(); + session + .input_flow_files + .push(MockFlowFile::with_content(b"hello")); + + processor + .trigger(&mut context, &mut session, &logger) + .expect("trigger should succeed"); + + let transferred = session.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 2); + + let original = transferred + .iter() + .find(|t| t.relationship == ORIGINAL.name) + .expect("an original FlowFile should be transferred"); + let enrichment = transferred + .iter() + .find(|t| t.relationship == ENRICHMENT.name) + .expect("an enrichment FlowFile should be transferred"); + + assert_eq!( + original + .flow_file + .attributes + .get(FORK_ROLE_ATTR.name) + .unwrap(), + "ORIGINAL" + ); + assert_eq!( + enrichment + .flow_file + .attributes + .get(FORK_ROLE_ATTR.name) + .unwrap(), + "ENRICHMENT" + ); + + let orig_group = original + .flow_file + .attributes + .get(GROUP_ID_ATTR.name) + .unwrap(); + let enr_group = enrichment + .flow_file + .attributes + .get(GROUP_ID_ATTR.name) + .unwrap(); + assert_eq!(orig_group, enr_group); + assert!(!orig_group.is_empty()); + } +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment/fork_enrichment_def.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment/fork_enrichment_def.rs new file mode 100644 index 000000000..ed887b94e --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment/fork_enrichment_def.rs @@ -0,0 +1,27 @@ +use crate::processors::attributes::{FORK_ROLE_ATTR, GROUP_ID_ATTR}; +use crate::processors::fork_enrichment::ForkEnrichment; +use minifi_native::{ + property_definitions, OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, + PropertyDefinition, Relationship, +}; + +pub(crate) const ORIGINAL: Relationship = Relationship { + name: "original", + description: "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes.", +}; + +pub(crate) const ENRICHMENT: Relationship = Relationship { + name: "enrichment", + description: "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes.", +}; + +impl ProcessorDefinition for ForkEnrichment { + const DESCRIPTION: &'static str = "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' r [...] + const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required; + const SUPPORTS_DYNAMIC_PROPERTIES: bool = false; + const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; + const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[FORK_ROLE_ATTR, GROUP_ID_ATTR]; + const RELATIONSHIPS: &'static [Relationship] = &[ORIGINAL, ENRICHMENT]; + + const PROPERTIES: &[PropertyDefinition] = property_definitions![]; +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes.rs new file mode 100644 index 000000000..defc19a60 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes.rs @@ -0,0 +1,339 @@ +use std::time::{Duration, Instant}; + +use minifi_native::macros::ComponentIdentifier; +use minifi_native::{ + FlowFileStore, GetProperty, Logger, MinifiError, MutTrigger, OnTriggerResult, ProcessContext, + ProcessSession, Schedule, warn, +}; + +use crate::processors::attributes::{FORK_ROLE_ATTR, GROUP_ID_ATTR, JOIN_ROLE_ATTR, Role}; +use crate::processors::join_enrichment_attributes::join_enrichment_attributes_def::{ + BATCH_SIZE, INVALID, JOINED, ORIGINAL, TIMEOUT_PROP, TIMEOUT_REL, +}; + +mod join_enrichment_attributes_def; + +#[derive(ComponentIdentifier)] +pub(crate) struct JoinEnrichmentAttributes { + batch_size: usize, + timeout: Option<Duration>, + pending: FlowFileStore<String>, +} + +fn get_role<Session: ProcessSession>( + session: &Session, + flow_file: &Session::FlowFile, +) -> Option<Role> { + session + .get_attribute(flow_file, FORK_ROLE_ATTR.name)? + .parse() + .ok() +} + +fn get_role_and_group_id<Session: ProcessSession>( + flow_file: &Session::FlowFile, + session: &Session, +) -> Option<(Role, String)> { + let role = get_role(session, flow_file)?; + let group_id = session + .get_required_attribute(flow_file, GROUP_ID_ATTR.name) + .ok()?; + Some((role, group_id)) +} + +fn join<Session: ProcessSession>( + session: &mut Session, + original_ff: Session::FlowFile, + enrichment_ff: Session::FlowFile, +) -> Result<(), MinifiError> { + let mut joined_ff = session.clone_ff(&original_ff)?; + + let mut enrichment_attrs: Vec<(String, String)> = Vec::new(); + session.for_each_attribute(&enrichment_ff, |key, value| { + enrichment_attrs.push((key.to_string(), value.to_string())); + }); + for (key, value) in &enrichment_attrs { + session.set_attribute(&mut joined_ff, key, value)?; + } + + session.set_attribute(&mut joined_ff, JOIN_ROLE_ATTR.name, "JOINED")?; + + session.transfer(original_ff, ORIGINAL.name)?; + session.transfer(enrichment_ff, ORIGINAL.name)?; + session.transfer(joined_ff, JOINED.name)?; + Ok(()) +} + +impl JoinEnrichmentAttributes { + fn handle_flow_file<Session>( + &mut self, + incoming: Session::FlowFile, + session: &mut Session, + role: Role, + group_id: String, + ) -> Result<(), MinifiError> + where + Session: ProcessSession, + { + let Some(pending) = self.pending.take(session, &group_id)? else { + return self.pending.store(session, group_id, incoming); + }; + + match (role, get_role(session, &pending)) { + (Role::Original, Some(Role::Enrichment)) => join(session, incoming, pending), + (Role::Enrichment, Some(Role::Original)) => join(session, pending, incoming), + _ => { + session.transfer(incoming, INVALID.name)?; + session.transfer(pending, INVALID.name)?; + Ok(()) + } + } + } +} + +impl Schedule for JoinEnrichmentAttributes { + fn schedule<Ctx: GetProperty, L: Logger>( + context: &Ctx, + _logger: &L, + ) -> Result<Self, MinifiError> + where + Self: Sized, + { + Ok(Self { + batch_size: context.get_property(&BATCH_SIZE)?.unwrap_or(usize::MAX), + timeout: context + .get_property(&TIMEOUT_PROP)? + .filter(|d| !d.is_zero()), + pending: FlowFileStore::new(), + }) + } +} + +impl JoinEnrichmentAttributes { + fn on_trigger<Session, Lggr>( + &mut self, + session: &mut Session, + logger: &Lggr, + now: Instant, + ) -> Result<OnTriggerResult, MinifiError> + where + Session: ProcessSession, + Lggr: Logger, + { + for _ in 0..self.batch_size { + let Some(flow_file) = session.get() else { + break; + }; + + let Some((role, group_id)) = get_role_and_group_id(&flow_file, session) else { + warn!(logger, "Missing required attribute"); + session.transfer(flow_file, INVALID.name)?; + continue; + }; + + self.handle_flow_file(flow_file, session, role, group_id)?; + } + + if let Some(timeout) = self.timeout { + for timed_out in self.pending.drain_expired(session, now, timeout)? { + session.transfer(timed_out, TIMEOUT_REL.name)?; + } + } + + Ok(OnTriggerResult::Ok) + } +} + +impl MutTrigger for JoinEnrichmentAttributes { + fn trigger<Context, Session, Lggr>( + &mut self, + _context: &mut Context, + session: &mut Session, + logger: &Lggr, + ) -> Result<OnTriggerResult, MinifiError> + where + Context: ProcessContext, + Session: ProcessSession<FlowFile = Context::FlowFile>, + Lggr: Logger, + { + self.on_trigger(session, logger, Instant::now()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use minifi_native::{MockFlowFile, MockLogger, MockProcessContext, MockProcessSession}; + + fn flow_file(role: &str, group_id: &str, content: &[u8]) -> MockFlowFile { + let mut ff = MockFlowFile::with_content(content); + ff.attributes + .insert(FORK_ROLE_ATTR.name.to_string(), role.to_string()); + ff.attributes + .insert(GROUP_ID_ATTR.name.to_string(), group_id.to_string()); + ff + } + + fn scheduled() -> (JoinEnrichmentAttributes, MockProcessContext, MockLogger) { + let logger = MockLogger::new(); + let context = MockProcessContext::new(); + let processor = JoinEnrichmentAttributes::schedule(&context, &logger).unwrap(); + (processor, context, logger) + } + + fn with_timeout(timeout: Duration) -> JoinEnrichmentAttributes { + JoinEnrichmentAttributes { + batch_size: usize::MAX, + timeout: Some(timeout), + pending: FlowFileStore::new(), + } + } + + #[test] + fn joins_original_and_enrichment_arriving_on_separate_triggers() { + let (mut processor, mut context, logger) = scheduled(); + + // Trigger 1: only the original arrives -> stored, nothing transferred. + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"original-content")); + processor + .trigger(&mut context, &mut session1, &logger) + .unwrap(); + assert_eq!(session1.num_of_transferred_flow_files(), 0); + + // Trigger 2: the enrichment arrives (with an extra attribute) -> join. + let mut enrichment = flow_file("ENRICHMENT", "g1", b"enrichment-content"); + enrichment + .attributes + .insert("enriched".to_string(), "yes".to_string()); + let mut session2 = MockProcessSession::new(); + session2.input_flow_files.push(enrichment); + processor + .trigger(&mut context, &mut session2, &logger) + .unwrap(); + + let transferred = session2.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 3); + assert_eq!( + transferred + .iter() + .filter(|t| t.relationship == ORIGINAL.name) + .count(), + 2 + ); + + let joined = transferred + .iter() + .find(|t| t.relationship == JOINED.name) + .expect("a joined FlowFile should be transferred"); + // Original's content is preserved. + assert!(joined.flow_file.content_eq("original-content")); + // Enrichment's attributes are merged in, and the role is set to JOINED. + assert_eq!(joined.flow_file.attributes.get("enriched").unwrap(), "yes"); + assert_eq!( + joined + .flow_file + .attributes + .get(JOIN_ROLE_ATTR.name) + .unwrap(), + "JOINED" + ); + } + + #[test] + fn missing_required_attribute_routes_to_invalid() { + let (mut processor, mut context, logger) = scheduled(); + + let mut session = MockProcessSession::new(); + session + .input_flow_files + .push(MockFlowFile::with_content(b"no attributes")); + processor + .trigger(&mut context, &mut session, &logger) + .unwrap(); + + let transferred = session.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 1); + assert_eq!(transferred[0].relationship, INVALID.name); + } + + #[test] + fn duplicate_role_for_a_group_routes_both_to_invalid() { + let (mut processor, mut context, logger) = scheduled(); + + // Trigger 1: first original stored. + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"first")); + processor + .trigger(&mut context, &mut session1, &logger) + .unwrap(); + assert_eq!(session1.num_of_transferred_flow_files(), 0); + + // Trigger 2: a second original for the same group -> both invalid. + let mut session2 = MockProcessSession::new(); + session2 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"second")); + processor + .trigger(&mut context, &mut session2, &logger) + .unwrap(); + + let transferred = session2.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 2); + assert!(transferred.iter().all(|t| t.relationship == INVALID.name)); + } + + #[test] + fn unpaired_half_is_routed_to_timeout_after_the_timeout_elapses() { + let logger = MockLogger::new(); + let mut processor = with_timeout(Duration::from_secs(60)); + let start = Instant::now(); + + // Trigger 1: the original arrives and is stored; nothing is emitted yet. + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"lonely")); + processor.on_trigger(&mut session1, &logger, start).unwrap(); + assert_eq!(session1.num_of_transferred_flow_files(), 0); + + // Trigger 2, long past the timeout, with no pair: the stored half is + // handed back and routed to `timeout`. + let mut session2 = MockProcessSession::new(); + processor + .on_trigger(&mut session2, &logger, start + Duration::from_secs(120)) + .unwrap(); + + let transferred = session2.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 1); + assert_eq!(transferred[0].relationship, TIMEOUT_REL.name); + assert!(transferred[0].flow_file.content_eq("lonely")); + assert!(processor.pending.is_empty()); + } + + #[test] + fn half_within_the_timeout_window_is_kept() { + let logger = MockLogger::new(); + let mut processor = with_timeout(Duration::from_secs(60)); + let start = Instant::now(); + + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"waiting")); + processor.on_trigger(&mut session1, &logger, start).unwrap(); + + // A later trigger still inside the timeout window: nothing is emitted and + // the half is still pending its pair. + let mut session2 = MockProcessSession::new(); + processor + .on_trigger(&mut session2, &logger, start + Duration::from_secs(1)) + .unwrap(); + assert_eq!(session2.num_of_transferred_flow_files(), 0); + assert_eq!(processor.pending.len(), 1); + } +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes/join_enrichment_attributes_def.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes/join_enrichment_attributes_def.rs new file mode 100644 index 000000000..732dee0af --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes/join_enrichment_attributes_def.rs @@ -0,0 +1,48 @@ +use crate::processors::attributes::JOIN_ROLE_ATTR; +use crate::processors::join_enrichment_attributes::JoinEnrichmentAttributes; +use minifi_native::{ + OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, Property, PropertyDefinition, + Relationship, property_definitions, +}; +use std::time::Duration; + +pub(crate) const INVALID: Relationship = Relationship { + name: "invalid", + description: "Any FlowFiles without the requisite attributes will be routed here", +}; + +pub(crate) const JOINED: Relationship = Relationship { + name: "joined", + description: "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship", +}; + +pub(crate) const ORIGINAL: Relationship = Relationship { + name: "original", + description: "Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles.", +}; + +pub(crate) const TIMEOUT_REL: Relationship = Relationship { + name: "timeout", + description: "If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship.", +}; + +pub(crate) const BATCH_SIZE: Property<Option<usize>> = Property::new( + "Batch Size", + "The maximum number of FlowFiles to process in each trigger", +); + +pub(crate) const TIMEOUT_PROP: Property<Option<Duration>> = Property::new( + "Timeout", + "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship.", +); + +impl ProcessorDefinition for JoinEnrichmentAttributes { + const DESCRIPTION: &'static str = "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's)."; + const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required; + const SUPPORTS_DYNAMIC_PROPERTIES: bool = false; + const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; + const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[JOIN_ROLE_ATTR]; + const RELATIONSHIPS: &'static [Relationship] = &[INVALID, JOINED, ORIGINAL, TIMEOUT_REL]; + + const PROPERTIES: &[PropertyDefinition] = property_definitions![BATCH_SIZE, TIMEOUT_PROP]; +} diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/mod.rs similarity index 62% copy from minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs copy to minifi_rust/extensions/minifi_enrichment/src/processors/mod.rs index 580c04626..c55651fe4 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/mod.rs @@ -15,25 +15,6 @@ // specific language governing permissions and limitations // under the License. -use crate::api::FlowFile; -use minifi_native_sys::minifi_flow_file; - -pub struct CffiFlowFile<'a> { - ptr: *mut minifi_flow_file, - _lifetime: std::marker::PhantomData<&'a ()>, -} - -impl CffiFlowFile<'_> { - pub(crate) fn new(ptr: *mut minifi_flow_file) -> Self { - Self { - ptr, - _lifetime: std::marker::PhantomData, - } - } - - pub(crate) fn get_ptr(&self) -> *mut minifi_flow_file { - self.ptr - } -} - -impl FlowFile for CffiFlowFile<'_> {} +pub(crate) mod attributes; +pub(crate) mod fork_enrichment; +pub(crate) mod join_enrichment_attributes; diff --git a/minifi_rust/minifi_native/src/api.rs b/minifi_rust/minifi_native/src/api.rs index 49f3d88d5..b9be4db56 100644 --- a/minifi_rust/minifi_native/src/api.rs +++ b/minifi_rust/minifi_native/src/api.rs @@ -20,6 +20,7 @@ pub(crate) mod component_definition_traits; pub(crate) mod controller_service; pub(crate) mod errors; mod flow_file; +pub(crate) mod flow_file_store; pub(crate) mod logger; mod process_context; pub(crate) mod process_session; @@ -32,6 +33,7 @@ pub(crate) mod raw_processor; pub(crate) mod relationship; pub use flow_file::{FlowFile, GetId}; +pub use flow_file_store::FlowFileStore; pub use logger::{LogLevel, Logger}; pub use process_context::ProcessContext; pub use process_session::{InputStream, OutputStream, ProcessSession}; diff --git a/minifi_rust/minifi_native/src/api/flow_file_store.rs b/minifi_rust/minifi_native/src/api/flow_file_store.rs new file mode 100644 index 000000000..47d002f64 --- /dev/null +++ b/minifi_rust/minifi_native/src/api/flow_file_store.rs @@ -0,0 +1,194 @@ +// 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 +// +// https://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. + +use std::any::Any; +use std::collections::HashMap; +use std::hash::Hash; +use std::time::{Duration, Instant}; + +use crate::MinifiError; +use crate::api::ProcessSession; + +struct StoredEntry { + flow_file: Box<dyn Any + Send>, + stored_at: Instant, +} + +pub struct FlowFileStore<K> { + stored: HashMap<K, StoredEntry>, +} + +impl<K: Eq + Hash> FlowFileStore<K> { + pub fn new() -> Self { + Self { + stored: HashMap::new(), + } + } + + pub fn store<S>( + &mut self, + session: &mut S, + key: K, + flow_file: S::FlowFile, + ) -> Result<(), MinifiError> + where + S: ProcessSession, + { + let stashed = session.stash(flow_file)?; + self.stored.insert( + key, + StoredEntry { + flow_file: Box::new(stashed), + stored_at: Instant::now(), + }, + ); + Ok(()) + } + + pub fn take<S>(&mut self, session: &mut S, key: &K) -> Result<Option<S::FlowFile>, MinifiError> + where + S: ProcessSession, + { + let Some(entry) = self.stored.remove(key) else { + return Ok(None); + }; + Ok(Some(Self::add_back(session, entry)?)) + } + + pub fn drain_expired<S>( + &mut self, + session: &mut S, + now: Instant, + max_age: Duration, + ) -> Result<Vec<S::FlowFile>, MinifiError> + where + S: ProcessSession, + K: Clone, + { + let expired_keys: Vec<K> = self + .stored + .iter() + .filter(|(_, entry)| now.saturating_duration_since(entry.stored_at) > max_age) + .map(|(key, _)| key.clone()) + .collect(); + + let mut expired = Vec::with_capacity(expired_keys.len()); + for key in expired_keys { + let entry = self.stored.remove(&key).expect("key came from the map"); + expired.push(Self::add_back(session, entry)?); + } + Ok(expired) + } + + fn add_back<S>(session: &mut S, entry: StoredEntry) -> Result<S::FlowFile, MinifiError> + where + S: ProcessSession, + { + let stashed = *entry + .flow_file + .downcast::<S::StashedFlowFile>() + .expect("stashed FlowFile type is stable within a processor instance"); + session.unstash(stashed) + } + + pub fn contains(&self, key: &K) -> bool { + self.stored.contains_key(key) + } + + pub fn len(&self) -> usize { + self.stored.len() + } + + pub fn is_empty(&self) -> bool { + self.stored.is_empty() + } +} + +impl<K: Eq + Hash> Default for FlowFileStore<K> { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::MockFlowFile; + use crate::mock::MockProcessSession; + + #[test] + fn store_then_take_round_trips_a_flow_file() { + let mut session = MockProcessSession::new(); + let mut store: FlowFileStore<String> = FlowFileStore::new(); + + let mut ff = MockFlowFile::new(); + ff.attributes + .insert("enrichment.group.id".to_string(), "g1".to_string()); + let id = ff.id.clone(); + + store + .store(&mut session, "g1".to_string(), ff) + .expect("store should succeed"); + assert_eq!(store.len(), 1); + assert!(store.contains(&"g1".to_string())); + + let taken = store + .take(&mut session, &"g1".to_string()) + .expect("take should succeed") + .expect("a flow file should be stored under g1"); + assert_eq!(taken.id, id); + assert!(store.is_empty()); + } + + #[test] + fn take_missing_key_returns_none() { + let mut session = MockProcessSession::new(); + let mut store: FlowFileStore<String> = FlowFileStore::new(); + + let result = store + .take(&mut session, &"missing".to_string()) + .expect("take should not error"); + assert!(result.is_none()); + } + + #[test] + fn drain_expired_only_returns_entries_older_than_max_age() { + let mut session = MockProcessSession::new(); + let mut store: FlowFileStore<String> = FlowFileStore::new(); + + store + .store(&mut session, "g1".to_string(), MockFlowFile::new()) + .expect("store should succeed"); + + let max_age = Duration::from_secs(60); + // `now` right after storing: nothing has aged past max_age. + let just_now = Instant::now(); + let drained = store + .drain_expired(&mut session, just_now, max_age) + .expect("drain should succeed"); + assert!(drained.is_empty()); + assert_eq!(store.len(), 1); + + // `now` far in the future: the entry is older than max_age and comes back. + let future = just_now + Duration::from_secs(120); + let drained = store + .drain_expired(&mut session, future, max_age) + .expect("drain should succeed"); + assert_eq!(drained.len(), 1); + assert!(store.is_empty()); + } +} diff --git a/minifi_rust/minifi_native/src/api/process_session.rs b/minifi_rust/minifi_native/src/api/process_session.rs index eb121d3bf..383eb8109 100644 --- a/minifi_rust/minifi_native/src/api/process_session.rs +++ b/minifi_rust/minifi_native/src/api/process_session.rs @@ -31,11 +31,24 @@ pub enum IoState { pub trait ProcessSession { type FlowFile: FlowFile; + /// A handle to a flow file stashed with the session, storable across triggers. + /// + /// Produced by [`stash`](Self::stash) and consumed by [`unstash`](Self::unstash). + /// Unlike [`Self::FlowFile`] (which is bound to the session), this type is + /// `'static` so it can be held between `onTrigger` invocations - see + /// [`FlowFileStore`](crate::FlowFileStore). + type StashedFlowFile: 'static + Send; + fn create(&mut self) -> Result<Self::FlowFile, MinifiError>; fn get(&mut self) -> Option<Self::FlowFile>; + fn clone_ff(&mut self, flow_file: &Self::FlowFile) -> Result<Self::FlowFile, MinifiError>; fn transfer(&self, flow_file: Self::FlowFile, relationship: &str) -> Result<(), MinifiError>; fn remove(&mut self, flow_file: Self::FlowFile) -> Result<(), MinifiError>; + fn stash(&mut self, flow_file: Self::FlowFile) -> Result<Self::StashedFlowFile, MinifiError>; + + fn unstash(&mut self, stashed: Self::StashedFlowFile) -> Result<Self::FlowFile, MinifiError>; + fn set_attribute( &self, flow_file: &mut Self::FlowFile, @@ -66,4 +79,15 @@ pub trait ProcessSession { F: FnOnce(&mut dyn InputStream) -> Result<R, MinifiError>; fn get_flow_file_id(&self, flow_file: &Self::FlowFile) -> Result<String, MinifiError>; + + fn get_required_attribute( + &self, + flow_file: &Self::FlowFile, + attr_key: &str, + ) -> Result<String, MinifiError> { + self.get_attribute(flow_file, attr_key) + .ok_or(MinifiError::MissingRequiredAttribute( + attr_key.to_owned().into(), + )) + } } diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs index 580c04626..9b1a9b012 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs @@ -16,7 +16,7 @@ // under the License. use crate::api::FlowFile; -use minifi_native_sys::minifi_flow_file; +use minifi_native_sys::{minifi_flow_file, minifi_stashed_flow_file}; pub struct CffiFlowFile<'a> { ptr: *mut minifi_flow_file, @@ -37,3 +37,33 @@ impl CffiFlowFile<'_> { } impl FlowFile for CffiFlowFile<'_> {} + +/// A handle to a flow file that has been stashed with a session and can be kept +/// across `onTrigger` invocations (see [`ProcessSession::stash`]). +/// +/// Unlike [`CffiFlowFile`] it carries no session lifetime, so it is `'static`. +/// The stashed flow file itself is owned by the processor/session on the minifi +/// side; this is only a token used to [`unstash`](crate::ProcessSession::unstash) +/// it again. There is deliberately no explicit free: a token dropped without +/// being unstashed simply leaves the stashed flow file in the processor, which +/// minifi reclaims when the processor is torn down. +/// +/// [`ProcessSession::stash`]: crate::ProcessSession::stash +pub struct CffiStashedFlowFile { + ptr: *mut minifi_stashed_flow_file, +} + +// SAFETY: the handle is only ever touched from the single processor instance +// that stashed it; the stashed flow file lives on the minifi side, not behind +// this pointer. +unsafe impl Send for CffiStashedFlowFile {} + +impl CffiStashedFlowFile { + pub(crate) fn new(ptr: *mut minifi_stashed_flow_file) -> Self { + Self { ptr } + } + + pub(crate) fn get_ptr(&self) -> *mut minifi_stashed_flow_file { + self.ptr + } +} diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs index 9bed92bd6..c1b4e2b69 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use super::c_ffi_flow_file::CffiFlowFile; +use super::c_ffi_flow_file::{CffiFlowFile, CffiStashedFlowFile}; use crate::MinifiError; use crate::api::process_session::{IoState, OutputStream}; use crate::api::{InputStream, ProcessSession}; @@ -24,12 +24,14 @@ use crate::c_ffi::c_ffi_streams::{CffiInputStream, CffiOutputStream}; use minifi_native_sys::{ minifi_input_stream, minifi_input_stream_read, minifi_input_stream_size, minifi_io_status_MINIFI_IO_CANCEL, minifi_io_status_MINIFI_IO_ERROR, minifi_output_stream, - minifi_output_stream_write, minifi_process_session, minifi_process_session_create, - minifi_process_session_get, minifi_process_session_get_flow_file_attribute, + minifi_output_stream_write, minifi_process_session, minifi_process_session_clone, + minifi_process_session_create, minifi_process_session_get, + minifi_process_session_get_flow_file_attribute, minifi_process_session_get_flow_file_attributes, minifi_process_session_get_flow_file_id, minifi_process_session_read, minifi_process_session_remove, - minifi_process_session_set_flow_file_attribute, minifi_process_session_transfer, - minifi_process_session_write, minifi_status_MINIFI_STATUS_SUCCESS, minifi_string_view, + minifi_process_session_set_flow_file_attribute, minifi_process_session_stash, + minifi_process_session_transfer, minifi_process_session_unstash, minifi_process_session_write, + minifi_stashed_flow_file, minifi_status_MINIFI_STATUS_SUCCESS, minifi_string_view, }; use std::ffi::{CString, c_void}; use std::io::Read; @@ -163,6 +165,7 @@ impl<'a> CffiProcessSession<'a> { impl<'a> ProcessSession for CffiProcessSession<'a> { type FlowFile = CffiFlowFile<'a>; // FlowFile shouldn't outlive the Session + type StashedFlowFile = CffiStashedFlowFile; // detached token, storable across triggers fn create(&mut self) -> Result<Self::FlowFile, MinifiError> { let ff_ptr = unsafe { minifi_process_session_create(self.ptr, std::ptr::null_mut()) }; @@ -182,6 +185,15 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { } } + fn clone_ff(&mut self, flow_file: &Self::FlowFile) -> Result<Self::FlowFile, MinifiError> { + let ff_ptr = unsafe { minifi_process_session_clone(self.ptr, flow_file.get_ptr()) }; + if ff_ptr.is_null() { + Err(MinifiError::UnknownError) + } else { + Ok(CffiFlowFile::new(ff_ptr)) + } + } + fn transfer(&self, flow_file: Self::FlowFile, relationship: &str) -> Result<(), MinifiError> { let c_relationship = CString::new(relationship)?; unsafe { @@ -216,6 +228,34 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { } } + fn stash(&mut self, flow_file: Self::FlowFile) -> Result<Self::StashedFlowFile, MinifiError> { + let mut stashed: *mut minifi_stashed_flow_file = std::ptr::null_mut(); + unsafe { + match minifi_process_session_stash(self.ptr, flow_file.get_ptr(), &mut stashed) { + #[allow(non_upper_case_globals)] + minifi_status_MINIFI_STATUS_SUCCESS => Ok(CffiStashedFlowFile::new(stashed)), + err_code => Err(MinifiError::StatusError(( + "minifi_process_session_stash".into(), + NonZeroU32::new_unchecked(err_code), // SAFETY we checked against 0 (minifi_status_MINIFI_STATUS_SUCCESS) + ))), + } + } + } + + fn unstash(&mut self, stashed: Self::StashedFlowFile) -> Result<Self::FlowFile, MinifiError> { + let mut flow_file: *mut minifi_native_sys::minifi_flow_file = std::ptr::null_mut(); + unsafe { + match minifi_process_session_unstash(self.ptr, stashed.get_ptr(), &mut flow_file) { + #[allow(non_upper_case_globals)] + minifi_status_MINIFI_STATUS_SUCCESS => Ok(CffiFlowFile::new(flow_file)), + err_code => Err(MinifiError::StatusError(( + "minifi_process_session_unstash".into(), + NonZeroU32::new_unchecked(err_code), // SAFETY we checked against 0 (minifi_status_MINIFI_STATUS_SUCCESS) + ))), + } + } + } + fn set_attribute( &self, flow_file: &mut Self::FlowFile, diff --git a/minifi_rust/minifi_native/src/lib.rs b/minifi_rust/minifi_native/src/lib.rs index bbef73995..7b0b242ff 100644 --- a/minifi_rust/minifi_native/src/lib.rs +++ b/minifi_rust/minifi_native/src/lib.rs @@ -58,9 +58,9 @@ pub use api::process_session::IoState; pub use api::attribute::{GetAttribute, OutputAttribute}; pub use api::{ - DataSize, FlowFile, GetId, InputStream, NonBlankPath, OnTriggerResult, OutputStream, - ProcessContext, ProcessSession, ProcessorInputRequirement, PropertyConstraints, PropertyType, - Relationship, StandardPropertyValidator, + DataSize, FlowFile, FlowFileStore, GetId, InputStream, NonBlankPath, OnTriggerResult, + OutputStream, ProcessContext, ProcessSession, ProcessorInputRequirement, PropertyConstraints, + PropertyType, Relationship, StandardPropertyValidator, }; pub use minifi_native_macros as macros; diff --git a/minifi_rust/minifi_native/src/mock/mock_flow_file.rs b/minifi_rust/minifi_native/src/mock/mock_flow_file.rs index c998335a2..8abd11de2 100644 --- a/minifi_rust/minifi_native/src/mock/mock_flow_file.rs +++ b/minifi_rust/minifi_native/src/mock/mock_flow_file.rs @@ -57,6 +57,16 @@ impl MockFlowFile { } } + /// A deep copy with the same content and attributes but a fresh id, + /// mirroring `ProcessSession::clone_ff`. + pub fn deep_clone(&self) -> MockFlowFile { + Self { + content: RefCell::new(self.content.borrow().clone()), + attributes: self.attributes.clone(), + id: next_mock_flow_file_id(), + } + } + pub fn content_len(&self) -> usize { self.content.borrow().len() } diff --git a/minifi_rust/minifi_native/src/mock/mock_process_session.rs b/minifi_rust/minifi_native/src/mock/mock_process_session.rs index b43f582cf..29ac4d394 100644 --- a/minifi_rust/minifi_native/src/mock/mock_process_session.rs +++ b/minifi_rust/minifi_native/src/mock/mock_process_session.rs @@ -34,6 +34,7 @@ pub struct MockProcessSession { impl ProcessSession for MockProcessSession { type FlowFile = MockFlowFile; + type StashedFlowFile = MockFlowFile; fn create(&mut self) -> Result<Self::FlowFile, MinifiError> { Ok(Self::FlowFile::new()) @@ -41,6 +42,11 @@ impl ProcessSession for MockProcessSession { fn get(&mut self) -> Option<Self::FlowFile> { self.input_flow_files.pop() } + + fn clone_ff(&mut self, flow_file: &Self::FlowFile) -> Result<Self::FlowFile, MinifiError> { + Ok(flow_file.deep_clone()) + } + fn transfer(&self, flow_file: Self::FlowFile, relationship: &str) -> Result<(), MinifiError> { self.transferred_flow_files .borrow_mut() @@ -55,6 +61,14 @@ impl ProcessSession for MockProcessSession { Ok(()) } + fn stash(&mut self, flow_file: Self::FlowFile) -> Result<Self::StashedFlowFile, MinifiError> { + Ok(flow_file) + } + + fn unstash(&mut self, stashed: Self::StashedFlowFile) -> Result<Self::FlowFile, MinifiError> { + Ok(stashed) + } + fn set_attribute( &self, flow_file: &mut Self::FlowFile,
