You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2026/08/13 09:36:25 UTC
(nifi-minifi-cpp) branch minifi_rust_impr_2 created (now a755e9a97)
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a change to branch minifi_rust_impr_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
at a755e9a97 change file structure
This branch includes the following new commits:
new 358dfef04 MINIFICPP-2886 Improve rust api trigger errors
new a755e9a97 change file structure
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(nifi-minifi-cpp) 02/02: change file structure
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch minifi_rust_impr_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit a755e9a974aeb6a7c18361bda5d877cd5670e3b6
Author: Martin Zink <ma...@apache.org>
AuthorDate: Wed Aug 12 21:04:13 2026 +0200
change file structure
---
.../src/processors/asciify_german.rs | 105 +++++++++-
.../asciify_german/processor_definition.rs | 35 ----
.../src/processors/asciify_german/relationships.rs | 28 ---
.../src/processors/asciify_german/tests.rs | 88 ---------
.../src/processors/generate_flow_file.rs | 160 +++++++++++++--
.../{properties.rs => definitions.rs} | 31 ++-
.../generate_flow_file/processor_definition.rs | 42 ----
.../processors/generate_flow_file/relationships.rs | 23 ---
.../src/processors/generate_flow_file/tests.rs | 155 ---------------
.../src/processors/get_file.rs | 215 ++++++++++++++++++---
.../get_file/{properties.rs => definitions.rs} | 48 ++++-
.../src/processors/get_file/output_attributes.rs | 30 ---
.../processors/get_file/processor_definition.rs | 51 -----
.../src/processors/get_file/relationships.rs | 23 ---
.../src/processors/get_file/tests.rs | 186 ------------------
.../src/processors/kamikaze_processor.rs | 135 +++++++++++--
.../kamikaze_processor/processor_definition.rs | 36 ----
.../processors/kamikaze_processor/properties.rs | 41 ----
.../processors/kamikaze_processor/relationships.rs | 23 ---
.../src/processors/kamikaze_processor/tests.rs | 97 ----------
.../src/processors/log_attribute.rs | 168 ++++++++++++++--
.../{properties.rs => definitions.rs} | 34 +++-
.../log_attribute/processor_definition.rs | 45 -----
.../src/processors/log_attribute/relationships.rs | 23 ---
.../src/processors/log_attribute/tests.rs | 160 ---------------
.../src/processors/lorem_ipsum_cs_user.rs | 70 +++++--
.../lorem_ipsum_cs_user/processor_definition.rs | 39 ----
.../processors/lorem_ipsum_cs_user/properties.rs | 34 ----
.../lorem_ipsum_cs_user/relationships.rs | 23 ---
.../src/processors/lorem_ipsum_cs_user/tests.rs | 36 ----
.../src/processors/put_file.rs | 170 ++++++++++++++--
.../src/processors/put_file/definitions.rs | 101 ++++++++++
.../processors/put_file/processor_definition.rs | 54 ------
.../src/processors/put_file/properties.rs | 41 ----
.../src/processors/put_file/relationships.rs | 28 ---
.../src/processors/put_file/tests.rs | 162 ----------------
.../processors/put_file/unix_only_properties.rs | 29 ---
37 files changed, 1135 insertions(+), 1634 deletions(-)
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs
index 6d67c7d05..28732984c 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs
@@ -17,14 +17,22 @@
// This processor is used to test streaming flow file transforms, with changing FlowFile sizes
-use crate::processors::asciify_german::relationships::FAILURE;
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
- FlowFileStreamTransform, GetProperty, InputStream, Logger, MinifiError, OutputStream,
- ProcessError, RouteErrorExt, Schedule, TransformStreamResult,
+ FlowFileStreamTransform, GetProperty, InputStream, Logger, MinifiError, OutputAttribute,
+ OutputStream, ProcessError, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
+ Relationship, RouteErrorExt, Schedule, TransformStreamResult,
};
-mod relationships;
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "All asciified flowfiles are routed here",
+};
+
+pub(crate) const FAILURE: Relationship = Relationship {
+ name: "failure",
+ description: "Non-german flowfiles are routed here",
+};
#[derive(Debug, ComponentIdentifier)]
pub(crate) struct AsciifyGerman {}
@@ -75,10 +83,93 @@ impl FlowFileStreamTransform for AsciifyGerman {
}
output_stream.flush()?;
- Ok(TransformStreamResult::new(&relationships::SUCCESS))
+ Ok(TransformStreamResult::new(&SUCCESS))
+ }
+}
+
+impl ProcessorDefinition for AsciifyGerman {
+ const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: This processor switches German characters with their ascii counterparts. (to test stream API)";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS, FAILURE];
+ fn properties() -> &'static [PropertyDefinition] {
+ &[]
}
}
-mod processor_definition;
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::*;
+ use minifi_native::{IoState, MockLogger, MockProcessContext};
+ use std::io::BufReader;
+
+ #[test]
+ fn schedule_succeeds_with_default_values() {
+ assert!(AsciifyGerman::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
+ }
+
+ #[test]
+ fn simple_test() {
+ let process_context = MockProcessContext::new();
+ let context = MockProcessContext::new();
+ let logger = MockLogger::new();
+
+ let asciify_german =
+ AsciifyGerman::schedule(&process_context, &logger).expect("Should succeed");
+ let input_str = "Falsches Üben von Xylophonmusik quält jeden größeren Zwerg.";
+ let mut input_stream = BufReader::new(input_str.as_bytes());
+ let mut output_vec: Vec<u8> = Vec::new();
+ {
+ let result = asciify_german
+ .transform(&context, &mut input_stream, &mut output_vec, &logger)
+ .expect("Should succeed");
+ assert_eq!(result.write_status(), IoState::Ok);
+ assert_eq!(result.target_relationship_name(), SUCCESS.name);
+ }
+ assert_eq!(
+ output_vec,
+ "Falsches Ueben von Xylophonmusik quaelt jeden groesseren Zwerg.".as_bytes()
+ );
+ }
+
+ #[test]
+ fn simple_failure_test() {
+ let process_context = MockProcessContext::new();
+ let context = MockProcessContext::new();
+ let logger = MockLogger::new();
+
+ let asciify_german =
+ AsciifyGerman::schedule(&process_context, &logger).expect("Should succeed");
+ let input_str = "Üldögélő műújságíró";
+ let mut input_stream = BufReader::new(input_str.as_bytes());
+ let mut output_vec: Vec<u8> = Vec::new();
+ {
+ let result = asciify_german
+ .transform(&context, &mut input_stream, &mut output_vec, &logger)
+ .expect("Should succeed");
+ assert_eq!(result.write_status(), IoState::Cancel);
+ assert_eq!(result.target_relationship_name(), FAILURE.name);
+ }
+ assert_eq!(output_vec, "Ueldoeg".as_bytes());
+ }
+
+ #[test]
+ fn truncated_umlaut_at_eof_routes_to_failure() {
+ let context = MockProcessContext::new();
+ let logger = MockLogger::new();
+
+ let asciify_german = AsciifyGerman::schedule(&context, &logger).expect("Should succeed");
+ // A lone 0xC3 is the leading byte of every German umlaut in UTF-8; on EOF the
+ // sequence is incomplete and the processor must report failure rather than
+ // silently truncating.
+ let input_bytes: &[u8] = &[b'a', 0xC3];
+ let mut input_stream = BufReader::new(input_bytes);
+ let mut output_vec: Vec<u8> = Vec::new();
+
+ let result =
+ asciify_german.transform(&context, &mut input_stream, &mut output_vec, &logger);
+ assert!(result.is_err());
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs
deleted file mode 100644
index 4baa9d9cc..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs
+++ /dev/null
@@ -1,35 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::processors::asciify_german::AsciifyGerman;
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship,
-};
-
-impl ProcessorDefinition for AsciifyGerman {
- const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: This processor switches German characters with their ascii counterparts. (to test stream API)";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
- const RELATIONSHIPS: &'static [Relationship] =
- &[super::relationships::SUCCESS, super::relationships::FAILURE];
- fn properties() -> &'static [PropertyDefinition] {
- &[]
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/relationships.rs
deleted file mode 100644
index e624aa3bc..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/relationships.rs
+++ /dev/null
@@ -1,28 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "All asciified flowfiles are routed here",
-};
-
-pub(crate) const FAILURE: Relationship = Relationship {
- name: "failure",
- description: "Non-german flowfiles are routed here",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs
deleted file mode 100644
index 1a2a77d60..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs
+++ /dev/null
@@ -1,88 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use crate::processors::asciify_german::relationships::SUCCESS;
-use minifi_native::{IoState, MockLogger, MockProcessContext};
-use std::io::BufReader;
-
-#[test]
-fn schedule_succeeds_with_default_values() {
- assert!(AsciifyGerman::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
-}
-
-#[test]
-fn simple_test() {
- let process_context = MockProcessContext::new();
- let context = MockProcessContext::new();
- let logger = MockLogger::new();
-
- let asciify_german =
- AsciifyGerman::schedule(&process_context, &logger).expect("Should succeed");
- let input_str = "Falsches Üben von Xylophonmusik quält jeden größeren Zwerg.";
- let mut input_stream = BufReader::new(input_str.as_bytes());
- let mut output_vec: Vec<u8> = Vec::new();
- {
- let result = asciify_german
- .transform(&context, &mut input_stream, &mut output_vec, &logger)
- .expect("Should succeed");
- assert_eq!(result.write_status(), IoState::Ok);
- assert_eq!(result.target_relationship_name(), SUCCESS.name);
- }
- assert_eq!(
- output_vec,
- "Falsches Ueben von Xylophonmusik quaelt jeden groesseren Zwerg.".as_bytes()
- );
-}
-
-#[test]
-fn simple_failure_test() {
- let process_context = MockProcessContext::new();
- let context = MockProcessContext::new();
- let logger = MockLogger::new();
-
- let asciify_german =
- AsciifyGerman::schedule(&process_context, &logger).expect("Should succeed");
- let input_str = "Üldögélő műújságíró";
- let mut input_stream = BufReader::new(input_str.as_bytes());
- let mut output_vec: Vec<u8> = Vec::new();
- {
- let result = asciify_german
- .transform(&context, &mut input_stream, &mut output_vec, &logger)
- .expect("Should succeed");
- assert_eq!(result.write_status(), IoState::Cancel);
- assert_eq!(result.target_relationship_name(), FAILURE.name);
- }
- assert_eq!(output_vec, "Ueldoeg".as_bytes());
-}
-
-#[test]
-fn truncated_umlaut_at_eof_routes_to_failure() {
- let context = MockProcessContext::new();
- let logger = MockLogger::new();
-
- let asciify_german = AsciifyGerman::schedule(&context, &logger).expect("Should succeed");
- // A lone 0xC3 is the leading byte of every German umlaut in UTF-8; on EOF the
- // sequence is incomplete and the processor must report failure rather than
- // silently truncating.
- let input_bytes: &[u8] = &[b'a', 0xC3];
- let mut input_stream = BufReader::new(input_bytes);
- let mut output_vec: Vec<u8> = Vec::new();
-
- let result = asciify_german.transform(&context, &mut input_stream, &mut output_vec, &logger);
- assert!(result.is_err());
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
index b5c9045e5..d1a29fd1c 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
@@ -27,8 +27,7 @@ use rand::distr::Alphanumeric;
use std::cmp::PartialEq;
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
-mod properties;
-mod relationships;
+mod definitions;
#[derive(
Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames, IntoStaticStr, PropertyType,
@@ -62,12 +61,12 @@ impl Schedule for GenerateFlowFileRs {
where
Self: Sized,
{
- let is_unique = context.get_property(&properties::UNIQUE_FLOW_FILES)?;
- let is_text = context.get_property(&properties::DATA_FORMAT)? == DataFormat::Text;
- let has_custom_text = context.get_property(&properties::CUSTOM_TEXT)?.is_some();
+ let is_unique = context.get_property(&definitions::UNIQUE_FLOW_FILES)?;
+ let is_text = context.get_property(&definitions::DATA_FORMAT)? == DataFormat::Text;
+ let has_custom_text = context.get_property(&definitions::CUSTOM_TEXT)?.is_some();
- let file_size = context.get_property(&properties::FILE_SIZE)?;
- let batch_size = context.get_property(&properties::BATCH_SIZE)?;
+ let file_size = context.get_property(&definitions::FILE_SIZE)?;
+ let batch_size = context.get_property(&definitions::BATCH_SIZE)?;
let mode = Self::get_mode(is_unique, is_text, has_custom_text, file_size);
let data_generated_during_on_schedule =
@@ -162,7 +161,7 @@ impl Trigger for GenerateFlowFileRs {
// flow files.
custom_text_for_batch = Some(
context
- .get_raw_property(&properties::CUSTOM_TEXT, None)?
+ .get_raw_property(&definitions::CUSTOM_TEXT, None)?
.ok_or_else(|| {
MinifiError::custom(
"GenerateFlowFile is in CustomText mode but the \"Custom Text\" \
@@ -186,13 +185,150 @@ impl Trigger for GenerateFlowFileRs {
session.write(&ff, non_unique_data_buffer)?;
}
}
- session.transfer(ff, relationships::SUCCESS.name)?;
+ session.transfer(ff, definitions::SUCCESS.name)?;
}
Ok(OnTriggerResult::Ok)
}
}
-pub(crate) mod processor_definition;
-
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::definitions::*;
+ use super::*;
+ use minifi_native::{MockLogger, MockProcessContext, MockProcessSession};
+
+ #[test]
+ fn schedule_succeeds_with_default_values() {
+ assert!(
+ GenerateFlowFileRs::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok()
+ );
+ }
+
+ #[test]
+ fn generate_flow_file_empty_test() {
+ let logger = MockLogger::new();
+ let mut context = MockProcessContext::new();
+ context
+ .properties
+ .insert(FILE_SIZE.name().to_string(), "0".to_string());
+ context
+ .properties
+ .insert(UNIQUE_FLOW_FILES.name().to_string(), "false".to_string());
+ context
+ .properties
+ .insert(DATA_FORMAT.name().to_string(), "Text".to_string());
+
+ let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
+ let mut session = MockProcessSession::new();
+ assert_eq!(
+ processor
+ .trigger(&mut context, &mut session, &logger)
+ .unwrap(),
+ OnTriggerResult::Ok
+ );
+ let result_flow_files = session.transferred_flow_files.borrow();
+ assert_eq!(result_flow_files.len(), 1);
+ assert_eq!(result_flow_files[0].flow_file.content_len(), 0);
+ }
+
+ #[test]
+ fn generate_custom_text() {
+ let mut context = MockProcessContext::new();
+ context
+ .properties
+ .insert(FILE_SIZE.name().to_string(), "0".to_string());
+ context
+ .properties
+ .insert(UNIQUE_FLOW_FILES.name().to_string(), "false".to_string());
+ context
+ .properties
+ .insert(DATA_FORMAT.name().to_string(), "Text".to_string());
+ context
+ .properties
+ .insert(CUSTOM_TEXT.name().to_string(), "foo bar baz".to_string());
+
+ let logger = MockLogger::new();
+ let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
+
+ let mut session = MockProcessSession::new();
+ assert_eq!(
+ processor
+ .trigger(&mut context, &mut session, &logger)
+ .expect("Should trigger successfully"),
+ OnTriggerResult::Ok
+ );
+ let result_flow_files = session.transferred_flow_files.borrow();
+ assert_eq!(result_flow_files.len(), 1);
+ assert!(result_flow_files[0].flow_file.content_eq("foo bar baz"),);
+ }
+
+ #[test]
+ fn random_bytes_unique() {
+ let mut context = MockProcessContext::new();
+ context
+ .properties
+ .insert(FILE_SIZE.name().to_string(), "40 B".to_string());
+ context
+ .properties
+ .insert(UNIQUE_FLOW_FILES.name().to_string(), "true".to_string());
+ context
+ .properties
+ .insert(DATA_FORMAT.name().to_string(), "Binary".to_string());
+ context
+ .properties
+ .insert(BATCH_SIZE.name().to_string(), "2".to_string());
+
+ let logger = MockLogger::new();
+ let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
+ let mut session = MockProcessSession::new();
+ assert_eq!(
+ processor
+ .trigger(&mut context, &mut session, &logger)
+ .expect("Should trigger successfully"),
+ OnTriggerResult::Ok
+ );
+ let result_flow_files = session.transferred_flow_files.borrow();
+ assert_eq!(result_flow_files.len(), 2);
+ assert_eq!(result_flow_files[0].flow_file.content_len(), 40);
+ assert_eq!(result_flow_files[1].flow_file.content_len(), 40);
+ assert_ne!(
+ *result_flow_files[0].flow_file.content.borrow(),
+ *result_flow_files[1].flow_file.content.borrow()
+ );
+ }
+
+ #[test]
+ fn random_bytes_non_unique() {
+ let mut context = MockProcessContext::new();
+ context
+ .properties
+ .insert(FILE_SIZE.name().to_string(), "40 B".to_string());
+ context
+ .properties
+ .insert(UNIQUE_FLOW_FILES.name().to_string(), "false".to_string());
+ context
+ .properties
+ .insert(DATA_FORMAT.name().to_string(), "Binary".to_string());
+ context
+ .properties
+ .insert(BATCH_SIZE.name().to_string(), "2".to_string());
+
+ let logger = MockLogger::new();
+ let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
+ let mut session = MockProcessSession::new();
+ assert_eq!(
+ processor
+ .trigger(&mut context, &mut session, &logger)
+ .expect("Should trigger successfully"),
+ OnTriggerResult::Ok
+ );
+ let result_flow_files = session.transferred_flow_files.borrow();
+ assert_eq!(result_flow_files.len(), 2);
+ assert_eq!(result_flow_files[0].flow_file.content_len(), 40);
+ assert_eq!(result_flow_files[1].flow_file.content_len(), 40);
+ assert_eq!(
+ *result_flow_files[0].flow_file.content.borrow(),
+ *result_flow_files[1].flow_file.content.borrow()
+ );
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/definitions.rs
similarity index 64%
rename from minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/properties.rs
rename to minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/definitions.rs
index d6f62767f..4708dcfce 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/properties.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/definitions.rs
@@ -15,8 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use super::DataFormat;
-use minifi_native::{DataSize, Property};
+use super::*;
+use minifi_native::{
+ DataSize, OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, Property,
+ PropertyDefinition, Relationship, property_definitions,
+};
+
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "success",
+};
pub(crate) const FILE_SIZE: Property<DataSize> =
Property::new("File Size", "The size of the file that will be used")
@@ -46,3 +54,22 @@ pub(crate) const CUSTOM_TEXT: Property<Option<String>> = Property::new(
"If Data Format is text and if Unique FlowFiles is false, then this custom text will be used as content of the generated FlowFiles and the File Size will be ignored. Finally, if Expression Language is used, evaluation will be performed only once per batch of generated FlowFiles",
)
.supports_expression_language();
+
+impl ProcessorDefinition for GenerateFlowFileRs {
+ const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: This processor creates FlowFiles with random data or custom content. GenerateFlowFile is useful for load testing, configuration, and simulation.";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Forbidden;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ FILE_SIZE,
+ BATCH_SIZE,
+ DATA_FORMAT,
+ UNIQUE_FLOW_FILES,
+ CUSTOM_TEXT
+ ];
+ PROPERTIES
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
deleted file mode 100644
index a3e0e8a68..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
+++ /dev/null
@@ -1,42 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::properties::*;
-use super::{GenerateFlowFileRs, relationships};
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, property_definitions,
-};
-
-impl ProcessorDefinition for GenerateFlowFileRs {
- const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: This processor creates FlowFiles with random data or custom content. GenerateFlowFile is useful for load testing, configuration, and simulation.";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Forbidden;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
- const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- fn properties() -> &'static [PropertyDefinition] {
- const PROPERTIES: &[PropertyDefinition] = property_definitions![
- FILE_SIZE,
- BATCH_SIZE,
- DATA_FORMAT,
- UNIQUE_FLOW_FILES,
- CUSTOM_TEXT
- ];
- PROPERTIES
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/relationships.rs
deleted file mode 100644
index 81fd6f904..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/relationships.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "success",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/tests.rs
deleted file mode 100644
index 710ed98dc..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/tests.rs
+++ /dev/null
@@ -1,155 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use crate::processors::generate_flow_file::properties::{
- BATCH_SIZE, CUSTOM_TEXT, DATA_FORMAT, UNIQUE_FLOW_FILES,
-};
-use minifi_native::{MockLogger, MockProcessContext, MockProcessSession};
-
-#[test]
-fn schedule_succeeds_with_default_values() {
- assert!(GenerateFlowFileRs::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
-}
-
-#[test]
-fn generate_flow_file_empty_test() {
- let logger = MockLogger::new();
- let mut context = MockProcessContext::new();
- context
- .properties
- .insert(properties::FILE_SIZE.name().to_string(), "0".to_string());
- context
- .properties
- .insert(UNIQUE_FLOW_FILES.name().to_string(), "false".to_string());
- context
- .properties
- .insert(DATA_FORMAT.name().to_string(), "Text".to_string());
-
- let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
- let mut session = MockProcessSession::new();
- assert_eq!(
- processor
- .trigger(&mut context, &mut session, &logger)
- .unwrap(),
- OnTriggerResult::Ok
- );
- let result_flow_files = session.transferred_flow_files.borrow();
- assert_eq!(result_flow_files.len(), 1);
- assert_eq!(result_flow_files[0].flow_file.content_len(), 0);
-}
-
-#[test]
-fn generate_custom_text() {
- let mut context = MockProcessContext::new();
- context
- .properties
- .insert(properties::FILE_SIZE.name().to_string(), "0".to_string());
- context
- .properties
- .insert(UNIQUE_FLOW_FILES.name().to_string(), "false".to_string());
- context
- .properties
- .insert(DATA_FORMAT.name().to_string(), "Text".to_string());
- context
- .properties
- .insert(CUSTOM_TEXT.name().to_string(), "foo bar baz".to_string());
-
- let logger = MockLogger::new();
- let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
-
- let mut session = MockProcessSession::new();
- assert_eq!(
- processor
- .trigger(&mut context, &mut session, &logger)
- .expect("Should trigger successfully"),
- OnTriggerResult::Ok
- );
- let result_flow_files = session.transferred_flow_files.borrow();
- assert_eq!(result_flow_files.len(), 1);
- assert!(result_flow_files[0].flow_file.content_eq("foo bar baz"),);
-}
-
-#[test]
-fn random_bytes_unique() {
- let mut context = MockProcessContext::new();
- context
- .properties
- .insert(properties::FILE_SIZE.name().to_string(), "40 B".to_string());
- context
- .properties
- .insert(UNIQUE_FLOW_FILES.name().to_string(), "true".to_string());
- context
- .properties
- .insert(DATA_FORMAT.name().to_string(), "Binary".to_string());
- context
- .properties
- .insert(BATCH_SIZE.name().to_string(), "2".to_string());
-
- let logger = MockLogger::new();
- let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
- let mut session = MockProcessSession::new();
- assert_eq!(
- processor
- .trigger(&mut context, &mut session, &logger)
- .expect("Should trigger successfully"),
- OnTriggerResult::Ok
- );
- let result_flow_files = session.transferred_flow_files.borrow();
- assert_eq!(result_flow_files.len(), 2);
- assert_eq!(result_flow_files[0].flow_file.content_len(), 40);
- assert_eq!(result_flow_files[1].flow_file.content_len(), 40);
- assert_ne!(
- *result_flow_files[0].flow_file.content.borrow(),
- *result_flow_files[1].flow_file.content.borrow()
- );
-}
-
-#[test]
-fn random_bytes_non_unique() {
- let mut context = MockProcessContext::new();
- context
- .properties
- .insert(properties::FILE_SIZE.name().to_string(), "40 B".to_string());
- context
- .properties
- .insert(UNIQUE_FLOW_FILES.name().to_string(), "false".to_string());
- context
- .properties
- .insert(DATA_FORMAT.name().to_string(), "Binary".to_string());
- context
- .properties
- .insert(BATCH_SIZE.name().to_string(), "2".to_string());
-
- let logger = MockLogger::new();
- let processor = GenerateFlowFileRs::schedule(&context, &logger).unwrap();
- let mut session = MockProcessSession::new();
- assert_eq!(
- processor
- .trigger(&mut context, &mut session, &logger)
- .expect("Should trigger successfully"),
- OnTriggerResult::Ok
- );
- let result_flow_files = session.transferred_flow_files.borrow();
- assert_eq!(result_flow_files.len(), 2);
- assert_eq!(result_flow_files[0].flow_file.content_len(), 40);
- assert_eq!(result_flow_files[1].flow_file.content_len(), 40);
- assert_eq!(
- *result_flow_files[0].flow_file.content.borrow(),
- *result_flow_files[1].flow_file.content.borrow()
- );
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
index 7ec9f4130..dd89424c2 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
@@ -17,13 +17,6 @@
// This is the (not production ready) reimplementation of the already existing standard GetFile processor
-use crate::processors::get_file::output_attributes::{
- ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE,
-};
-use crate::processors::get_file::properties::{
- BATCH_SIZE, DIRECTORY, IGNORE_HIDDEN_FILES, KEEP_SOURCE_FILE, MAX_AGE, MAX_SIZE, MIN_AGE,
- MIN_SIZE, RECURSE,
-};
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
@@ -37,8 +30,7 @@ use std::sync::Mutex;
use std::time::{Duration, Instant, SystemTime};
use walkdir::{DirEntry, WalkDir};
-mod properties;
-mod relationships;
+mod definitions;
#[derive(Debug)]
struct GetFileMetrics {
@@ -185,13 +177,17 @@ impl GetFileRs {
.expect("Successful FlowFile creation is expected");
if let Some(file_name) = path.file_name().and_then(|f| f.to_str()) {
- session.set_attribute(&mut ff, FILENAME_OUTPUT_ATTRIBUTE.name, file_name)?;
+ session.set_attribute(
+ &mut ff,
+ definitions::FILENAME_OUTPUT_ATTRIBUTE.name,
+ file_name,
+ )?;
} else {
warn!(logger, "Couldnt get filename of {:?}", path);
}
session.set_attribute(
&mut ff,
- ABSOLUTE_PATH_OUTPUT_ATTRIBUTE.name,
+ definitions::ABSOLUTE_PATH_OUTPUT_ATTRIBUTE.name,
path.to_string_lossy().trim(),
)?;
@@ -205,7 +201,7 @@ impl GetFileRs {
{
warn!(logger, "Failed to remove source file {:?}", err);
}
- session.transfer(ff, relationships::SUCCESS.name)?;
+ session.transfer(ff, definitions::SUCCESS.name)?;
Ok(())
}
@@ -223,7 +219,7 @@ impl Schedule for GetFileRs {
where
Self: Sized,
{
- let input_directory = context.get_property(&DIRECTORY)?;
+ let input_directory = context.get_property(&definitions::DIRECTORY)?;
if !input_directory.is_dir() {
return Err(MinifiError::custom(format!(
"{:?} is not a valid directory",
@@ -231,17 +227,17 @@ impl Schedule for GetFileRs {
)));
}
- let recursive = context.get_property(&RECURSE)?;
+ let recursive = context.get_property(&definitions::RECURSE)?;
- let keep_source_file = context.get_property(&KEEP_SOURCE_FILE)?;
+ let keep_source_file = context.get_property(&definitions::KEEP_SOURCE_FILE)?;
- let poll_interval = context.get_property(&properties::POLLING_INTERVAL)?;
- let min_size = context.get_property(&MIN_SIZE)?;
- let max_size = context.get_property(&MAX_SIZE)?;
- let min_age = context.get_property(&MIN_AGE)?;
- let max_age = context.get_property(&MAX_AGE)?;
- let batch_size = context.get_property(&BATCH_SIZE)?;
- let ignore_hidden_files = context.get_property(&IGNORE_HIDDEN_FILES)?;
+ let poll_interval = context.get_property(&definitions::POLLING_INTERVAL)?;
+ let min_size = context.get_property(&definitions::MIN_SIZE)?;
+ let max_size = context.get_property(&definitions::MAX_SIZE)?;
+ let min_age = context.get_property(&definitions::MIN_AGE)?;
+ let max_age = context.get_property(&definitions::MAX_AGE)?;
+ let batch_size = context.get_property(&definitions::BATCH_SIZE)?;
+ let ignore_hidden_files = context.get_property(&definitions::IGNORE_HIDDEN_FILES)?;
Ok(GetFileRs {
recursive,
@@ -311,8 +307,175 @@ impl Trigger for GetFileRs {
}
}
-pub(crate) mod processor_definition;
-
-mod output_attributes;
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::definitions::*;
+ use super::*;
+ use filetime::FileTime;
+ use minifi_native::{MockLogger, MockProcessContext, MockProcessSession};
+ use tempfile::TempDir;
+
+ #[test]
+ fn schedule_fails_without_input_dir() {
+ assert!(matches!(
+ GetFileRs::schedule(&MockProcessContext::new(), &MockLogger::new())
+ .err()
+ .unwrap(),
+ MinifiError::MissingRequiredProperty(_)
+ ));
+ }
+
+ #[test]
+ fn schedule_fails_with_invalid_input_dir() {
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ "Input Directory".to_string(),
+ "/invalid_directory".to_string(),
+ );
+ assert!(GetFileRs::schedule(&context, &MockLogger::new()).is_err());
+ }
+
+ #[test]
+ fn simple_get_file_test() {
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing GetFile");
+ let file_path = temp_dir.path().join("input_file");
+ std::fs::write(&file_path, "test").unwrap();
+
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ "Input Directory".to_string(),
+ temp_dir.path().to_str().unwrap().to_string(),
+ );
+
+ let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
+
+ let mut session = MockProcessSession::new();
+ get_file
+ .trigger(&mut context, &mut session, &MockLogger::new())
+ .expect("Should succeed");
+ assert_eq!(session.num_of_transferred_flow_files(), 1);
+ }
+
+ fn make_file(temp_dir: &TempDir, file_name: &str, size: usize, age: Duration) {
+ let path = temp_dir.path().join(file_name);
+ std::fs::write(&path, "a".repeat(size)).unwrap();
+ let file_time = FileTime::from_system_time(SystemTime::now() - age);
+ filetime::set_file_mtime(path, file_time).expect("Cannot set file time");
+ }
+
+ fn create_test_directory() -> TempDir {
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing GetFile");
+ make_file(&temp_dir, "small_new", 10, Duration::from_secs(10));
+ make_file(&temp_dir, "small_old", 11, Duration::from_secs(3600));
+
+ make_file(&temp_dir, "large_new", 1000, Duration::from_secs(0));
+ make_file(&temp_dir, "large_old", 2000, Duration::from_secs(3600));
+ make_file(&temp_dir, ".small_hidden", 10, Duration::from_secs(0));
+ temp_dir
+ }
+
+ #[test]
+ fn complex_dir_without_filters() {
+ let test_directory = create_test_directory();
+
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ "Input Directory".to_string(),
+ test_directory.path().to_str().unwrap().to_string(),
+ );
+ context
+ .properties
+ .insert("Batch Size".to_string(), "10".to_string());
+
+ let mut session = MockProcessSession::new();
+ let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
+ get_file
+ .trigger(&mut context, &mut session, &MockLogger::new())
+ .expect("Should succeed");
+ assert_eq!(session.num_of_transferred_flow_files(), 4);
+ }
+
+ fn test_complex_dir_with_filter(
+ property_name: &str,
+ property_vale: &str,
+ expected_filename_part: &str,
+ ) {
+ let test_directory = create_test_directory();
+
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ DIRECTORY.name().to_string(),
+ test_directory.path().to_str().unwrap().to_string(),
+ );
+ context
+ .properties
+ .insert(BATCH_SIZE.name().to_string(), "10".to_string());
+
+ context
+ .properties
+ .insert(property_name.to_string(), property_vale.to_string());
+
+ let mut session = MockProcessSession::new();
+ let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
+ get_file
+ .trigger(&mut context, &mut session, &MockLogger::new())
+ .expect("Should succeed");
+ assert_eq!(session.num_of_transferred_flow_files(), 2);
+ let transferred_flow_files = session.transferred_flow_files.borrow();
+ assert!(transferred_flow_files.iter().all(|transfer| {
+ transfer.relationship == SUCCESS.name
+ && transfer
+ .flow_file
+ .attributes
+ .get("filename")
+ .map(|filename| filename.contains(expected_filename_part))
+ .unwrap_or(false)
+ }));
+ let sum_file_len = transferred_flow_files
+ .iter()
+ .fold(0, |acc, transfer| acc + transfer.flow_file.content_len());
+
+ let metrics = get_file.calculate_metrics();
+ assert_eq!(metrics.len(), 2);
+ assert_eq!(metrics[0].0, "accepted_files".to_string());
+ assert_eq!(metrics[0].1, 2.0);
+ assert_eq!(metrics[1].0, "input_bytes".to_string());
+ assert_eq!(metrics[1].1, sum_file_len as f64);
+ }
+
+ #[test]
+ fn complex_dir_with_filters() {
+ test_complex_dir_with_filter(MIN_AGE.name(), "5 min", "old");
+ test_complex_dir_with_filter(MAX_AGE.name(), "5 min", "new");
+ test_complex_dir_with_filter(MIN_SIZE.name(), "50 B", "large");
+ test_complex_dir_with_filter(MAX_SIZE.name(), "50 B", "small");
+ }
+
+ #[test]
+ fn test_hidden_files_and_batch_size() {
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing GetFile");
+ make_file(&temp_dir, ".one", 10, Duration::from_secs(0));
+ make_file(&temp_dir, ".two", 10, Duration::from_secs(0));
+ make_file(&temp_dir, ".three", 10, Duration::from_secs(0));
+
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ DIRECTORY.name().to_string(),
+ temp_dir.path().to_str().unwrap().to_string(),
+ );
+ context
+ .properties
+ .insert(BATCH_SIZE.name().to_string(), "2".to_string());
+
+ context
+ .properties
+ .insert(IGNORE_HIDDEN_FILES.name().to_string(), "false".to_string());
+
+ let mut session = MockProcessSession::new();
+ let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
+ get_file
+ .trigger(&mut context, &mut session, &MockLogger::new())
+ .expect("Should succeed");
+ assert_eq!(session.num_of_transferred_flow_files(), 2);
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/definitions.rs
similarity index 59%
rename from minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/properties.rs
rename to minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/definitions.rs
index 39f6f60b6..76e07c675 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/properties.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/definitions.rs
@@ -15,9 +15,30 @@
// specific language governing permissions and limitations
// under the License.
-use minifi_native::{DataSize, NonBlankPath, Property};
+use super::*;
+use minifi_native::{
+ DataSize, NonBlankPath, OutputAttribute, ProcessorDefinition, ProcessorInputRequirement,
+ Property, PropertyDefinition, Relationship, property_definitions,
+};
use std::time::Duration;
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "The created FlowFiles are transferred here",
+};
+
+pub(crate) const FILENAME_OUTPUT_ATTRIBUTE: OutputAttribute = OutputAttribute {
+ name: "filename",
+ relationships: &["success"],
+ description: "The filename is set to the name of the file on disk",
+};
+
+pub(crate) const ABSOLUTE_PATH_OUTPUT_ATTRIBUTE: OutputAttribute = OutputAttribute {
+ name: "absolute.path",
+ relationships: &["success"],
+ description: "The full/absolute path from where a file was picked up. The current 'path' attribute is still populated, but may be a relative path",
+};
+
pub(crate) const DIRECTORY: Property<NonBlankPath> = Property::new(
"Input Directory",
"The input directory from which to pull files",
@@ -72,3 +93,28 @@ pub(crate) const BATCH_SIZE: Property<u64> = Property::new(
"The maximum number of files to pull in each iteration",
)
.with_default("10");
+
+impl ProcessorDefinition for GetFileRs {
+ const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: Creates FlowFiles from files in a directory. MiNiFi will ignore files for which it doesn't have read permissions.";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Forbidden;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] =
+ &[ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ DIRECTORY,
+ POLLING_INTERVAL,
+ RECURSE,
+ KEEP_SOURCE_FILE,
+ MIN_AGE,
+ MAX_AGE,
+ MIN_SIZE,
+ MAX_SIZE,
+ IGNORE_HIDDEN_FILES,
+ BATCH_SIZE,
+ ];
+ PROPERTIES
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/output_attributes.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/output_attributes.rs
deleted file mode 100644
index 1e6e63299..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/output_attributes.rs
+++ /dev/null
@@ -1,30 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::OutputAttribute;
-
-pub(crate) const FILENAME_OUTPUT_ATTRIBUTE: OutputAttribute = OutputAttribute {
- name: "filename",
- relationships: &["success"],
- description: "The filename is set to the name of the file on disk",
-};
-
-pub(crate) const ABSOLUTE_PATH_OUTPUT_ATTRIBUTE: OutputAttribute = OutputAttribute {
- name: "absolute.path",
- relationships: &["success"],
- description: "The full/absolute path from where a file was picked up. The current 'path' attribute is still populated, but may be a relative path",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
deleted file mode 100644
index 13dbbbbdc..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
+++ /dev/null
@@ -1,51 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::processors::get_file::output_attributes::{
- ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE,
-};
-use crate::processors::get_file::properties::*;
-use crate::processors::get_file::{GetFileRs, relationships};
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, property_definitions,
-};
-
-impl ProcessorDefinition for GetFileRs {
- const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: Creates FlowFiles from files in a directory. MiNiFi will ignore files for which it doesn't have read permissions.";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Forbidden;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] =
- &[ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE];
- const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- fn properties() -> &'static [PropertyDefinition] {
- const PROPERTIES: &[PropertyDefinition] = property_definitions![
- DIRECTORY,
- POLLING_INTERVAL,
- RECURSE,
- KEEP_SOURCE_FILE,
- MIN_AGE,
- MAX_AGE,
- MIN_SIZE,
- MAX_SIZE,
- IGNORE_HIDDEN_FILES,
- BATCH_SIZE,
- ];
- PROPERTIES
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/relationships.rs
deleted file mode 100644
index 48207d0ad..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/relationships.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "The created FlowFiles are transferred here",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
deleted file mode 100644
index d7b9ed96f..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
+++ /dev/null
@@ -1,186 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use crate::processors::get_file::relationships::SUCCESS;
-use filetime::FileTime;
-use minifi_native::{MockLogger, MockProcessContext, MockProcessSession};
-use tempfile::TempDir;
-
-#[test]
-fn schedule_fails_without_input_dir() {
- assert!(matches!(
- GetFileRs::schedule(&MockProcessContext::new(), &MockLogger::new())
- .err()
- .unwrap(),
- MinifiError::MissingRequiredProperty(_)
- ));
-}
-
-#[test]
-fn schedule_fails_with_invalid_input_dir() {
- let mut context = MockProcessContext::new();
- context.properties.insert(
- "Input Directory".to_string(),
- "/invalid_directory".to_string(),
- );
- assert!(GetFileRs::schedule(&context, &MockLogger::new()).is_err());
-}
-
-#[test]
-fn simple_get_file_test() {
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing GetFile");
- let file_path = temp_dir.path().join("input_file");
- std::fs::write(&file_path, "test").unwrap();
-
- let mut context = MockProcessContext::new();
- context.properties.insert(
- "Input Directory".to_string(),
- temp_dir.path().to_str().unwrap().to_string(),
- );
-
- let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
-
- let mut session = MockProcessSession::new();
- get_file
- .trigger(&mut context, &mut session, &MockLogger::new())
- .expect("Should succeed");
- assert_eq!(session.num_of_transferred_flow_files(), 1);
-}
-
-fn make_file(temp_dir: &TempDir, file_name: &str, size: usize, age: Duration) {
- let path = temp_dir.path().join(file_name);
- std::fs::write(&path, "a".repeat(size)).unwrap();
- let file_time = FileTime::from_system_time(SystemTime::now() - age);
- filetime::set_file_mtime(path, file_time).expect("Cannot set file time");
-}
-
-fn create_test_directory() -> TempDir {
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing GetFile");
- make_file(&temp_dir, "small_new", 10, Duration::from_secs(10));
- make_file(&temp_dir, "small_old", 11, Duration::from_secs(3600));
-
- make_file(&temp_dir, "large_new", 1000, Duration::from_secs(0));
- make_file(&temp_dir, "large_old", 2000, Duration::from_secs(3600));
- make_file(&temp_dir, ".small_hidden", 10, Duration::from_secs(0));
- temp_dir
-}
-
-#[test]
-fn complex_dir_without_filters() {
- let test_directory = create_test_directory();
-
- let mut context = MockProcessContext::new();
- context.properties.insert(
- "Input Directory".to_string(),
- test_directory.path().to_str().unwrap().to_string(),
- );
- context
- .properties
- .insert("Batch Size".to_string(), "10".to_string());
-
- let mut session = MockProcessSession::new();
- let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
- get_file
- .trigger(&mut context, &mut session, &MockLogger::new())
- .expect("Should succeed");
- assert_eq!(session.num_of_transferred_flow_files(), 4);
-}
-
-fn test_complex_dir_with_filter(
- property_name: &str,
- property_vale: &str,
- expected_filename_part: &str,
-) {
- let test_directory = create_test_directory();
-
- let mut context = MockProcessContext::new();
- context.properties.insert(
- DIRECTORY.name().to_string(),
- test_directory.path().to_str().unwrap().to_string(),
- );
- context
- .properties
- .insert(BATCH_SIZE.name().to_string(), "10".to_string());
-
- context
- .properties
- .insert(property_name.to_string(), property_vale.to_string());
-
- let mut session = MockProcessSession::new();
- let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
- get_file
- .trigger(&mut context, &mut session, &MockLogger::new())
- .expect("Should succeed");
- assert_eq!(session.num_of_transferred_flow_files(), 2);
- let transferred_flow_files = session.transferred_flow_files.borrow();
- assert!(transferred_flow_files.iter().all(|transfer| {
- transfer.relationship == SUCCESS.name
- && transfer
- .flow_file
- .attributes
- .get("filename")
- .map(|filename| filename.contains(expected_filename_part))
- .unwrap_or(false)
- }));
- let sum_file_len = transferred_flow_files
- .iter()
- .fold(0, |acc, transfer| acc + transfer.flow_file.content_len());
-
- let metrics = get_file.calculate_metrics();
- assert_eq!(metrics.len(), 2);
- assert_eq!(metrics[0].0, "accepted_files".to_string());
- assert_eq!(metrics[0].1, 2.0);
- assert_eq!(metrics[1].0, "input_bytes".to_string());
- assert_eq!(metrics[1].1, sum_file_len as f64);
-}
-
-#[test]
-fn complex_dir_with_filters() {
- test_complex_dir_with_filter(MIN_AGE.name(), "5 min", "old");
- test_complex_dir_with_filter(MAX_AGE.name(), "5 min", "new");
- test_complex_dir_with_filter(MIN_SIZE.name(), "50 B", "large");
- test_complex_dir_with_filter(MAX_SIZE.name(), "50 B", "small");
-}
-
-#[test]
-fn test_hidden_files_and_batch_size() {
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing GetFile");
- make_file(&temp_dir, ".one", 10, Duration::from_secs(0));
- make_file(&temp_dir, ".two", 10, Duration::from_secs(0));
- make_file(&temp_dir, ".three", 10, Duration::from_secs(0));
-
- let mut context = MockProcessContext::new();
- context.properties.insert(
- DIRECTORY.name().to_string(),
- temp_dir.path().to_str().unwrap().to_string(),
- );
- context
- .properties
- .insert(BATCH_SIZE.name().to_string(), "2".to_string());
-
- context
- .properties
- .insert(IGNORE_HIDDEN_FILES.name().to_string(), "false".to_string());
-
- let mut session = MockProcessSession::new();
- let get_file = GetFileRs::schedule(&context, &MockLogger::new()).unwrap();
- get_file
- .trigger(&mut context, &mut session, &MockLogger::new())
- .expect("Should succeed");
- assert_eq!(session.num_of_transferred_flow_files(), 2);
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
index 0faa04f18..4268d3b5c 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
@@ -17,16 +17,12 @@
// This processor is used to test Errors and panic during schedule/trigger
-mod properties;
-mod relationships;
-
-use crate::processors::kamikaze_processor::properties::{
- NOT_REGISTERED_PROPERTY, SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR, UNREGISTERED_CONTROLLER_SERVICE,
-};
+use crate::controller_services::lorem_ipsum_controller_service::LoremIpsumControllerService;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
- GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
- ProcessSession, Schedule, Trigger,
+ GetProperty, Logger, MinifiError, OnTriggerResult, OutputAttribute, ProcessContext,
+ ProcessError, ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property,
+ PropertyDefinition, Relationship, Schedule, Trigger, property_definitions,
};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -34,7 +30,7 @@ use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames, IntoStaticStr, PropertyType,
)]
#[strum(serialize_all = "PascalCase", const_into_str)]
-enum KamikazeBehaviour {
+pub(crate) enum KamikazeBehaviour {
ReturnErr,
ReturnOk,
GetNotRegisteredProperty,
@@ -42,6 +38,32 @@ enum KamikazeBehaviour {
Panic,
}
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "success relationship",
+};
+
+pub(crate) const SCHEDULE_BEHAVIOUR: Property<KamikazeBehaviour> = Property::new(
+ "Schedule Behaviour",
+ "What to do during the on_schedule method",
+)
+.with_default(KamikazeBehaviour::ReturnOk.into_str());
+
+pub(crate) const TRIGGER_BEHAVIOUR: Property<KamikazeBehaviour> =
+ Property::new("Trigger Behaviour", "What to do during the trigger method")
+ .with_default(KamikazeBehaviour::ReturnOk.into_str());
+
+pub(crate) const NOT_REGISTERED_PROPERTY: Property<Option<String>> = Property::new(
+ "Kamikaze Processor Property",
+ "Property purposely left out of Processor description",
+);
+
+pub(crate) const UNREGISTERED_CONTROLLER_SERVICE: Property<LoremIpsumControllerService> =
+ Property::new(
+ "Kamikaze Processor Property",
+ "Property purposely left out of Processor description",
+ );
+
#[derive(Debug, ComponentIdentifier)]
pub(crate) struct KamikazeProcessorRs {
trigger_behaviour: KamikazeBehaviour,
@@ -107,7 +129,98 @@ impl Trigger for KamikazeProcessorRs {
}
}
-pub(crate) mod processor_definition;
+impl ProcessorDefinition for KamikazeProcessorRs {
+ const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: This processor can fail or panic in on_trigger and on_schedule calls based on configuration. Only for testing purposes.";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Allowed;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR];
+ PROPERTIES
+ }
+}
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::*;
+ use minifi_native::{MockLogger, MockProcessContext, MockProcessSession, ProcessError};
+ use std::panic::AssertUnwindSafe;
+
+ #[test]
+ fn on_schedule_ok() {
+ let context = MockProcessContext::new();
+ let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new());
+ assert!(processor.is_ok());
+ }
+
+ #[test]
+ fn on_schedule_err() {
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ SCHEDULE_BEHAVIOUR.name().to_string(),
+ "ReturnErr".to_string(),
+ );
+ assert!(KamikazeProcessorRs::schedule(&context, &MockLogger::new()).is_err());
+ }
+
+ #[test]
+ fn on_schedule_panic() {
+ let mut context = MockProcessContext::new();
+ context
+ .properties
+ .insert(SCHEDULE_BEHAVIOUR.name().to_string(), "Panic".to_string());
+
+ let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
+ KamikazeProcessorRs::schedule(&context, &MockLogger::new())
+ }));
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn on_trigger_ok() {
+ let mut context = MockProcessContext::new();
+ let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()).unwrap();
+
+ let mut session = MockProcessSession::new();
+ assert_eq!(
+ processor
+ .trigger(&mut context, &mut session, &MockLogger::new())
+ .expect("Should trigger successfully"),
+ OnTriggerResult::Ok
+ );
+ }
+
+ #[test]
+ fn on_trigger_err() {
+ let mut context = MockProcessContext::new();
+ context.properties.insert(
+ TRIGGER_BEHAVIOUR.name().to_string(),
+ "ReturnErr".to_string(),
+ );
+ let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()).unwrap();
+
+ let mut session = MockProcessSession::new();
+ assert!(matches!(
+ processor.trigger(&mut context, &mut session, &MockLogger::new()),
+ Err(ProcessError::Fatal(MinifiError::CustomError(_)))
+ ));
+ }
+
+ #[test]
+ fn on_trigger_panic() {
+ let mut context = MockProcessContext::new();
+ context
+ .properties
+ .insert(TRIGGER_BEHAVIOUR.name().to_string(), "Panic".to_string());
+ let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()).unwrap();
+
+ let mut session = MockProcessSession::new();
+ let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
+ processor.trigger(&mut context, &mut session, &MockLogger::new())
+ }));
+ assert!(result.is_err());
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
deleted file mode 100644
index f9a47d115..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
+++ /dev/null
@@ -1,36 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, property_definitions,
-};
-
-impl ProcessorDefinition for KamikazeProcessorRs {
- const DESCRIPTION: &'static str = "RUST TEST PROCESSOR: This processor can fail or panic in on_trigger and on_schedule calls based on configuration. Only for testing purposes.";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Allowed;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
- const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- fn properties() -> &'static [PropertyDefinition] {
- const PROPERTIES: &[PropertyDefinition] =
- property_definitions![SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR];
- PROPERTIES
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/properties.rs
deleted file mode 100644
index 72f914536..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/properties.rs
+++ /dev/null
@@ -1,41 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::controller_services::lorem_ipsum_controller_service::LoremIpsumControllerService;
-use crate::processors::kamikaze_processor::KamikazeBehaviour;
-use minifi_native::Property;
-
-pub(crate) const SCHEDULE_BEHAVIOUR: Property<KamikazeBehaviour> = Property::new(
- "Schedule Behaviour",
- "What to do during the on_schedule method",
-)
-.with_default(KamikazeBehaviour::ReturnOk.into_str());
-
-pub(crate) const TRIGGER_BEHAVIOUR: Property<KamikazeBehaviour> =
- Property::new("Trigger Behaviour", "What to do during the trigger method")
- .with_default(KamikazeBehaviour::ReturnOk.into_str());
-
-pub(crate) const NOT_REGISTERED_PROPERTY: Property<Option<String>> = Property::new(
- "Kamikaze Processor Property",
- "Property purposely left out of Processor description",
-);
-
-pub(crate) const UNREGISTERED_CONTROLLER_SERVICE: Property<LoremIpsumControllerService> =
- Property::new(
- "Kamikaze Processor Property",
- "Property purposely left out of Processor description",
- );
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/relationships.rs
deleted file mode 100644
index 33e94d41a..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/relationships.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "success relationship",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
deleted file mode 100644
index 2b7f52166..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
+++ /dev/null
@@ -1,97 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use crate::processors::kamikaze_processor::properties::{SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR};
-use minifi_native::{MockLogger, MockProcessContext, MockProcessSession, ProcessError};
-use std::panic::AssertUnwindSafe;
-
-#[test]
-fn on_schedule_ok() {
- let context = MockProcessContext::new();
- let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new());
- assert!(processor.is_ok());
-}
-
-#[test]
-fn on_schedule_err() {
- let mut context = MockProcessContext::new();
- context.properties.insert(
- SCHEDULE_BEHAVIOUR.name().to_string(),
- "ReturnErr".to_string(),
- );
- let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new());
- assert!(matches!(processor, Err(MinifiError::CustomError(_))));
-}
-
-#[test]
-fn on_schedule_panic() {
- let mut context = MockProcessContext::new();
- context
- .properties
- .insert(SCHEDULE_BEHAVIOUR.name().to_string(), "Panic".to_string());
-
- let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
- KamikazeProcessorRs::schedule(&context, &MockLogger::new())
- }));
- assert!(result.is_err());
-}
-
-#[test]
-fn on_trigger_ok() {
- let mut context = MockProcessContext::new();
- let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()).unwrap();
-
- let mut session = MockProcessSession::new();
- assert_eq!(
- processor
- .trigger(&mut context, &mut session, &MockLogger::new())
- .expect("Should trigger successfully"),
- OnTriggerResult::Ok
- );
-}
-
-#[test]
-fn on_trigger_err() {
- let mut context = MockProcessContext::new();
- context.properties.insert(
- TRIGGER_BEHAVIOUR.name().to_string(),
- "ReturnErr".to_string(),
- );
- let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()).unwrap();
-
- let mut session = MockProcessSession::new();
- assert!(matches!(
- processor.trigger(&mut context, &mut session, &MockLogger::new()),
- Err(ProcessError::Fatal(MinifiError::CustomError(_)))
- ));
-}
-
-#[test]
-fn on_trigger_panic() {
- let mut context = MockProcessContext::new();
- context
- .properties
- .insert(TRIGGER_BEHAVIOUR.name().to_string(), "Panic".to_string());
- let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()).unwrap();
-
- let mut session = MockProcessSession::new();
- let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
- processor.trigger(&mut context, &mut session, &MockLogger::new())
- }));
- assert!(result.is_err());
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
index 329e281f4..c0a36d708 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
@@ -17,7 +17,6 @@
// This is the (not production ready) reimplementation of the already existing standard LogAttribute processor
-use crate::processors::log_attribute::properties::{FLOW_FILES_TO_LOG, LOG_LEVEL, LOG_PAYLOAD};
use minifi_native::StandardPropertyValidator::NonBlankValidator;
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
@@ -26,8 +25,7 @@ use minifi_native::{
log, trace,
};
-mod properties;
-mod relationships;
+mod definitions;
struct AttributeList {}
@@ -125,7 +123,7 @@ impl Trigger for LogAttributeRs {
if let Some(flow_file) = session.get() {
let log_msg = self.generate_log_message(session, &flow_file);
log!(logger, self.log_level, "{}", log_msg);
- session.transfer(flow_file, relationships::SUCCESS.name)?;
+ session.transfer(flow_file, definitions::SUCCESS.name)?;
flow_files_processed += 1;
} else {
break;
@@ -139,20 +137,20 @@ impl Trigger for LogAttributeRs {
impl Schedule for LogAttributeRs {
fn schedule<P: GetProperty, L>(context: &P, _logger: &L) -> Result<Self, MinifiError> {
- let log_level = context.get_property(&LOG_LEVEL)?;
- let log_payload = context.get_property(&LOG_PAYLOAD)?;
- let flow_files_to_log = context.get_property(&FLOW_FILES_TO_LOG)?;
- let attributes_to_log = context.get_property(&properties::ATTRIBUTES_TO_LOG)?;
- let attributes_to_ignore = context.get_property(&properties::ATTRIBUTES_TO_IGNORE)?;
+ let log_level = context.get_property(&definitions::LOG_LEVEL)?;
+ let log_payload = context.get_property(&definitions::LOG_PAYLOAD)?;
+ let flow_files_to_log = context.get_property(&definitions::FLOW_FILES_TO_LOG)?;
+ let attributes_to_log = context.get_property(&definitions::ATTRIBUTES_TO_LOG)?;
+ let attributes_to_ignore = context.get_property(&definitions::ATTRIBUTES_TO_IGNORE)?;
let dash_line = format!(
"{:-^50}",
context
- .get_property(&properties::LOG_PREFIX)?
+ .get_property(&definitions::LOG_PREFIX)?
.unwrap_or_default()
);
- let hex_encode_payload = context.get_property(&properties::HEX_ENCODE_PAYLOAD)?;
+ let hex_encode_payload = context.get_property(&definitions::HEX_ENCODE_PAYLOAD)?;
Ok(LogAttributeRs {
log_level,
@@ -166,7 +164,149 @@ impl Schedule for LogAttributeRs {
}
}
-pub(crate) mod processor_definition;
-
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::*;
+ use minifi_native::{
+ ComponentIdentifier, MockFlowFile, MockLogger, MockProcessContext, MockProcessSession,
+ };
+
+ #[test]
+ fn test_component_id() {
+ assert_eq!(
+ LogAttributeRs::CLASS_NAME,
+ "minifi_rs_playground::processors::log_attribute::LogAttributeRs"
+ );
+ assert_eq!(LogAttributeRs::GROUP_NAME, "minifi_rs_playground");
+ assert_eq!(LogAttributeRs::VERSION, "0.1.0");
+ }
+ #[test]
+ fn schedule_succeeds_with_default_values() {
+ assert!(LogAttributeRs::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
+ }
+
+ fn tester(
+ input_flow_files: Vec<MockFlowFile>,
+ log_level: LogLevel,
+ properties: Box<[(&str, &str)]>,
+ expected_log_msg: String,
+ ) {
+ let logger = MockLogger::new();
+ let mut context = MockProcessContext::new();
+ for (k, v) in properties {
+ context.properties.insert(k.to_string(), v.to_string());
+ }
+
+ let processor = LogAttributeRs::schedule(&context, &logger).unwrap();
+
+ let mut session = MockProcessSession::new();
+ for flow_file in input_flow_files {
+ session.input_flow_files.push(flow_file);
+ }
+ processor
+ .trigger(&mut context, &mut session, &logger)
+ .expect("The on_trigger should succeed");
+
+ let logs = logger.logs.lock().unwrap();
+ assert_eq!(logs[1], (log_level, expected_log_msg));
+ }
+
+ #[test]
+ fn warn_single_log_payload() {
+ let mut flow_file = MockFlowFile::with_content("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, [...]
+ flow_file
+ .attributes
+ .insert(String::from("apple"), String::from("apfel"));
+ let vec = vec![flow_file];
+
+ let expected =
+ "Logging for flow file
+--------------------------------------------------
+FlowFile Attributes Map Content
+key:apple value:apfel
+Payload:
+Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, velit elit euismod lacus, a placerat purus est a lacus. [...]
+--------------------------------------------------".to_string();
+ let properties_set = [
+ ("Log Payload", "true"),
+ ("Hexencode Payload", "false"),
+ ("Log Level", "Warn"),
+ ];
+ tester(vec, LogLevel::Warn, Box::new(properties_set), expected);
+ }
+
+ #[test]
+ fn critical_single_hexencode_payload() {
+ let mut flow_file = MockFlowFile::with_content("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, [...]
+ flow_file
+ .attributes
+ .insert(String::from("apple"), String::from("apfel"));
+ let vec = vec![flow_file];
+
+ let expected =
+ "Logging for flow file
+--------------------------------------------------
+FlowFile Attributes Map Content
+key:apple value:apfel
+Payload:
+4c6f72656d20697073756d20646f6c6f722073697420616d65742c20636f6e73656374657475722061646970697363696e6720656c69742e20496e746567657220666163696c69736973206469616d2073697420616d6574206e69736c20696e74657264756d2c20766974616520696e74657264756d206172637520766976657272612e204e616d20706c616365726174206d6920696e20657261742070656c6c656e7465737175652c20617420756c747269636573206f7263692066617563696275732e204372617320736f6c6c696369747564696e20696163756c697320706f73756572652e205365642074656d7075732c2064 [...]
+--------------------------------------------------".to_string();
+ let properties_set = [
+ ("Log Payload", "true"),
+ ("Hexencode Payload", "true"),
+ ("Log Level", "Critical"),
+ ];
+ tester(vec, LogLevel::Critical, Box::new(properties_set), expected);
+ }
+
+ #[test]
+ fn attributes_to_log_csv_trims_whitespace() {
+ let mut flow_file = MockFlowFile::with_content(b"payload");
+ flow_file
+ .attributes
+ .insert(String::from("apple"), String::from("apfel"));
+ flow_file
+ .attributes
+ .insert(String::from("pear"), String::from("birne"));
+ flow_file
+ .attributes
+ .insert(String::from("cherry"), String::from("kirsche"));
+ let vec = vec![flow_file];
+
+ // Value with spaces around commas: filter must match on trimmed names.
+ let expected = "Logging for flow file
+--------------------------------------------------
+FlowFile Attributes Map Content
+key:apple value:apfel
+key:pear value:birne
+--------------------------------------------------"
+ .to_string();
+ let properties_set = [
+ ("Log Payload", "false"),
+ ("Attributes to Log", "apple, pear"),
+ ];
+ tester(vec, LogLevel::Info, Box::new(properties_set), expected);
+ }
+
+ #[test]
+ fn default_level_multiple_attributes() {
+ let mut flow_file = MockFlowFile::with_content("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, [...]
+ flow_file
+ .attributes
+ .insert(String::from("apple"), String::from("apfel"));
+ flow_file
+ .attributes
+ .insert(String::from("pear"), String::from("birne"));
+ let vec = vec![flow_file];
+
+ let expected = "Logging for flow file
+--------------------------------------------------
+FlowFile Attributes Map Content
+key:apple value:apfel
+key:pear value:birne
+--------------------------------------------------"
+ .to_string();
+ let properties_set = [("Log Payload", "false")];
+ tester(vec, LogLevel::Info, Box::new(properties_set), expected);
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/definitions.rs
similarity index 66%
rename from minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/properties.rs
rename to minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/definitions.rs
index 38484faab..a7d255ee5 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/properties.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/definitions.rs
@@ -15,8 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use crate::processors::log_attribute::AttributeList;
-use minifi_native::{LogLevel, Property};
+use super::*;
+use minifi_native::{
+ LogLevel, OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, Property,
+ PropertyDefinition, Relationship, property_definitions,
+};
+
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "FlowFiles are transferred here after logging",
+};
pub(crate) const LOG_LEVEL: Property<LogLevel> = Property::new(
"Log Level",
@@ -56,3 +64,25 @@ pub(crate) const HEX_ENCODE_PAYLOAD: Property<bool> = Property::new(
"If true, the FlowFile's payload will be logged in a hexencoded format",
)
.with_default("false");
+
+impl ProcessorDefinition for LogAttributeRs {
+ const DESCRIPTION: &'static str =
+ "RUST TEST PROCESSOR: Logs attributes of flow files in the MiNiFi application log.";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ LOG_LEVEL,
+ ATTRIBUTES_TO_LOG,
+ ATTRIBUTES_TO_IGNORE,
+ LOG_PAYLOAD,
+ LOG_PREFIX,
+ FLOW_FILES_TO_LOG,
+ HEX_ENCODE_PAYLOAD,
+ ];
+ PROPERTIES
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
deleted file mode 100644
index 91e0b3fed..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
+++ /dev/null
@@ -1,45 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::processors::log_attribute::properties::*;
-use crate::processors::log_attribute::{LogAttributeRs, relationships};
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, property_definitions,
-};
-
-impl ProcessorDefinition for LogAttributeRs {
- const DESCRIPTION: &'static str =
- "RUST TEST PROCESSOR: Logs attributes of flow files in the MiNiFi application log.";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
- const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- fn properties() -> &'static [PropertyDefinition] {
- const PROPERTIES: &[PropertyDefinition] = property_definitions![
- LOG_LEVEL,
- ATTRIBUTES_TO_LOG,
- ATTRIBUTES_TO_IGNORE,
- LOG_PAYLOAD,
- LOG_PREFIX,
- FLOW_FILES_TO_LOG,
- HEX_ENCODE_PAYLOAD,
- ];
- PROPERTIES
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/relationships.rs
deleted file mode 100644
index b8ec9ae3c..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/relationships.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "FlowFiles are transferred here after logging",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/tests.rs
deleted file mode 100644
index 8c770fa6b..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/tests.rs
+++ /dev/null
@@ -1,160 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use minifi_native::{
- ComponentIdentifier, MockFlowFile, MockLogger, MockProcessContext, MockProcessSession,
-};
-
-#[test]
-fn test_component_id() {
- assert_eq!(
- LogAttributeRs::CLASS_NAME,
- "minifi_rs_playground::processors::log_attribute::LogAttributeRs"
- );
- assert_eq!(LogAttributeRs::GROUP_NAME, "minifi_rs_playground");
- assert_eq!(LogAttributeRs::VERSION, "0.1.0");
-}
-#[test]
-fn schedule_succeeds_with_default_values() {
- assert!(LogAttributeRs::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
-}
-
-fn tester(
- input_flow_files: Vec<MockFlowFile>,
- log_level: LogLevel,
- properties: Box<[(&str, &str)]>,
- expected_log_msg: String,
-) {
- let logger = MockLogger::new();
- let mut context = MockProcessContext::new();
- for (k, v) in properties {
- context.properties.insert(k.to_string(), v.to_string());
- }
-
- let processor = LogAttributeRs::schedule(&context, &logger).unwrap();
-
- let mut session = MockProcessSession::new();
- for flow_file in input_flow_files {
- session.input_flow_files.push(flow_file);
- }
- processor
- .trigger(&mut context, &mut session, &logger)
- .expect("The on_trigger should succeed");
-
- let logs = logger.logs.lock().unwrap();
- assert_eq!(logs[1], (log_level, expected_log_msg));
-}
-
-#[test]
-fn warn_single_log_payload() {
- let mut flow_file = MockFlowFile::with_content("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, veli [...]
- flow_file
- .attributes
- .insert(String::from("apple"), String::from("apfel"));
- let vec = vec![flow_file];
-
- let expected =
- "Logging for flow file
---------------------------------------------------
-FlowFile Attributes Map Content
-key:apple value:apfel
-Payload:
-Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, velit elit euismod lacus, a placerat purus est a lacus. [...]
---------------------------------------------------".to_string();
- let properties_set = [
- ("Log Payload", "true"),
- ("Hexencode Payload", "false"),
- ("Log Level", "Warn"),
- ];
- tester(vec, LogLevel::Warn, Box::new(properties_set), expected);
-}
-
-#[test]
-fn critical_single_hexencode_payload() {
- let mut flow_file = MockFlowFile::with_content("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, veli [...]
- flow_file
- .attributes
- .insert(String::from("apple"), String::from("apfel"));
- let vec = vec![flow_file];
-
- let expected =
- "Logging for flow file
---------------------------------------------------
-FlowFile Attributes Map Content
-key:apple value:apfel
-Payload:
-4c6f72656d20697073756d20646f6c6f722073697420616d65742c20636f6e73656374657475722061646970697363696e6720656c69742e20496e746567657220666163696c69736973206469616d2073697420616d6574206e69736c20696e74657264756d2c20766974616520696e74657264756d206172637520766976657272612e204e616d20706c616365726174206d6920696e20657261742070656c6c656e7465737175652c20617420756c747269636573206f7263692066617563696275732e204372617320736f6c6c696369747564696e20696163756c697320706f73756572652e205365642074656d7075732c2064 [...]
---------------------------------------------------".to_string();
- let properties_set = [
- ("Log Payload", "true"),
- ("Hexencode Payload", "true"),
- ("Log Level", "Critical"),
- ];
- tester(vec, LogLevel::Critical, Box::new(properties_set), expected);
-}
-
-#[test]
-fn attributes_to_log_csv_trims_whitespace() {
- let mut flow_file = MockFlowFile::with_content(b"payload");
- flow_file
- .attributes
- .insert(String::from("apple"), String::from("apfel"));
- flow_file
- .attributes
- .insert(String::from("pear"), String::from("birne"));
- flow_file
- .attributes
- .insert(String::from("cherry"), String::from("kirsche"));
- let vec = vec![flow_file];
-
- // Value with spaces around commas: filter must match on trimmed names.
- let expected = "Logging for flow file
---------------------------------------------------
-FlowFile Attributes Map Content
-key:apple value:apfel
-key:pear value:birne
---------------------------------------------------"
- .to_string();
- let properties_set = [
- ("Log Payload", "false"),
- ("Attributes to Log", "apple, pear"),
- ];
- tester(vec, LogLevel::Info, Box::new(properties_set), expected);
-}
-
-#[test]
-fn default_level_multiple_attributes() {
- let mut flow_file = MockFlowFile::with_content("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer facilisis diam sit amet nisl interdum, vitae interdum arcu viverra. Nam placerat mi in erat pellentesque, at ultrices orci faucibus. Cras sollicitudin iaculis posuere. Sed tempus, dolor nec lacinia suscipit, tellus odio venenatis odio, nec sollicitudin dolor augue non urna. Aliquam tincidunt viverra ipsum eget hendrerit. Suspendisse varius, augue vel fermentum varius, veli [...]
- flow_file
- .attributes
- .insert(String::from("apple"), String::from("apfel"));
- flow_file
- .attributes
- .insert(String::from("pear"), String::from("birne"));
- let vec = vec![flow_file];
-
- let expected = "Logging for flow file
---------------------------------------------------
-FlowFile Attributes Map Content
-key:apple value:apfel
-key:pear value:birne
---------------------------------------------------"
- .to_string();
- let properties_set = [("Log Payload", "false")];
- tester(vec, LogLevel::Info, Box::new(properties_set), expected);
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
index 4d32739e2..aa1a99723 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
@@ -16,16 +16,14 @@
// under the License.
// Simple test processor that uses a controller service
-mod properties;
-use crate::processors::lorem_ipsum_cs_user::properties::{
- CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE,
-};
-use crate::processors::lorem_ipsum_cs_user::relationships::SUCCESS;
+use crate::controller_services::dummy_controller_service::DummyControllerService;
+use crate::controller_services::lorem_ipsum_controller_service::LoremIpsumControllerService;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
Content, FlowFileSource, GeneratedFlowFile, GetControllerService, GetProperty, Logger,
- MinifiError, ProcessError, Schedule, trace,
+ MinifiError, OutputAttribute, ProcessError, ProcessorDefinition, ProcessorInputRequirement,
+ Property, PropertyDefinition, Relationship, Schedule, property_definitions, trace,
};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -33,11 +31,29 @@ use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames, IntoStaticStr, PropertyType,
)]
#[strum(serialize_all = "PascalCase", const_into_str)]
-enum WriteMethod {
+pub(crate) enum WriteMethod {
Buffer,
Stream,
}
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "All flowfile are routed here",
+};
+
+pub(crate) const CONTROLLER_SERVICE: Property<LoremIpsumControllerService> = Property::new(
+ "Lorem Ipsum Controller Service",
+ "Name of the lorem ipsum controller service",
+);
+
+pub(crate) const DUMMY_CONTROLLER_SERVICE: Property<Option<DummyControllerService>> = Property::new(
+ "Dummy Controller Service",
+ "Optional dummy controller service",
+);
+
+pub(crate) const WRITE_METHOD: Property<WriteMethod> =
+ Property::new("Write Method", "Which API to test").with_default(WriteMethod::Buffer.into_str());
+
#[derive(Debug, ComponentIdentifier)]
pub(crate) struct LoremIpsumCSUser {
write_method: WriteMethod,
@@ -48,7 +64,7 @@ impl Schedule for LoremIpsumCSUser {
where
Self: Sized,
{
- let write_method = context.get_property(&properties::WRITE_METHOD)?;
+ let write_method = context.get_property(&WRITE_METHOD)?;
Ok(Self { write_method })
}
}
@@ -84,8 +100,40 @@ impl FlowFileSource for LoremIpsumCSUser {
}
}
-pub(crate) mod processor_definition;
+impl ProcessorDefinition for LoremIpsumCSUser {
+ const DESCRIPTION: &'static str =
+ "RUST TEST PROCESSOR: Processor to test Controller Service API";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Forbidden;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD];
+ PROPERTIES
+ }
+}
-mod relationships;
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::*;
+ use minifi_native::{ComponentIdentifier, MockLogger, MockProcessContext};
+
+ #[test]
+ fn test_ids() {
+ assert_eq!(
+ LoremIpsumCSUser::CLASS_NAME,
+ "minifi_rs_playground::processors::lorem_ipsum_cs_user::LoremIpsumCSUser"
+ );
+ assert_eq!(LoremIpsumCSUser::GROUP_NAME, "minifi_rs_playground");
+ assert_eq!(LoremIpsumCSUser::VERSION, "0.1.0");
+ }
+
+ #[test]
+ fn schedules_with_controller() {
+ let context = MockProcessContext::new();
+ let schedule_result = LoremIpsumCSUser::schedule(&context, &MockLogger::new());
+ assert!(schedule_result.is_ok());
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
deleted file mode 100644
index a80605035..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
+++ /dev/null
@@ -1,39 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::LoremIpsumCSUser;
-use super::properties::*;
-use crate::processors::lorem_ipsum_cs_user::relationships::SUCCESS;
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, property_definitions,
-};
-
-impl ProcessorDefinition for LoremIpsumCSUser {
- const DESCRIPTION: &'static str =
- "RUST TEST PROCESSOR: Processor to test Controller Service API";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Forbidden;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
- const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
- fn properties() -> &'static [PropertyDefinition] {
- const PROPERTIES: &[PropertyDefinition] =
- property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD];
- PROPERTIES
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/properties.rs
deleted file mode 100644
index 0742c6fb0..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/properties.rs
+++ /dev/null
@@ -1,34 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::controller_services::dummy_controller_service::DummyControllerService;
-use crate::controller_services::lorem_ipsum_controller_service::LoremIpsumControllerService;
-use crate::processors::lorem_ipsum_cs_user::WriteMethod;
-use minifi_native::Property;
-
-pub(crate) const CONTROLLER_SERVICE: Property<LoremIpsumControllerService> = Property::new(
- "Lorem Ipsum Controller Service",
- "Name of the lorem ipsum controller service",
-);
-
-pub(crate) const DUMMY_CONTROLLER_SERVICE: Property<Option<DummyControllerService>> = Property::new(
- "Dummy Controller Service",
- "Optional dummy controller service",
-);
-
-pub(crate) const WRITE_METHOD: Property<WriteMethod> =
- Property::new("Write Method", "Which API to test").with_default(WriteMethod::Buffer.into_str());
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/relationships.rs
deleted file mode 100644
index 4c7c2d053..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/relationships.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "All flowfile are routed here",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/tests.rs
deleted file mode 100644
index 8a3ad0340..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/tests.rs
+++ /dev/null
@@ -1,36 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::processors::lorem_ipsum_cs_user::LoremIpsumCSUser;
-use minifi_native::{ComponentIdentifier, MockLogger, MockProcessContext, Schedule};
-
-#[test]
-fn test_ids() {
- assert_eq!(
- LoremIpsumCSUser::CLASS_NAME,
- "minifi_rs_playground::processors::lorem_ipsum_cs_user::LoremIpsumCSUser"
- );
- assert_eq!(LoremIpsumCSUser::GROUP_NAME, "minifi_rs_playground");
- assert_eq!(LoremIpsumCSUser::VERSION, "0.1.0");
-}
-
-#[test]
-fn schedules_with_controller() {
- let context = MockProcessContext::new();
- let schedule_result = LoremIpsumCSUser::schedule(&context, &MockLogger::new());
- assert!(schedule_result.is_ok());
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
index 1dd63ec2d..47292e04a 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
@@ -17,7 +17,7 @@
// This is the (not production ready) reimplementation of the already existing standard PutFile processor
-use crate::processors::put_file::relationships::{FAILURE, SUCCESS};
+use crate::processors::put_file::definitions::{FAILURE, SUCCESS};
use crate::processors::put_file::unix_permissions::PutFileUnixPermissions;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
@@ -28,11 +28,7 @@ use std::path::{Path, PathBuf};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
use walkdir::WalkDir;
-mod properties;
-mod relationships;
-#[cfg(unix)]
-mod unix_only_properties;
-
+mod definitions;
mod unix_permissions;
#[derive(
@@ -75,7 +71,7 @@ impl PutFileRs {
where
Ctx: GetProperty + GetAttribute + GetId,
{
- let directory = context.get_property(&properties::DIRECTORY)?;
+ let directory = context.get_property(&definitions::DIRECTORY)?;
let file_name = context
.get_attribute("filename")?
@@ -123,9 +119,8 @@ impl PutFileRs {
fn parse_unix_permissions<P: GetProperty>(
context: &P,
) -> Result<PutFileUnixPermissions, MinifiError> {
- let file_permissions = context.get_property(&unix_only_properties::PERMISSIONS)?;
- let directory_permissions =
- context.get_property(&unix_only_properties::DIRECTORY_PERMISSIONS)?;
+ let file_permissions = context.get_property(&definitions::PERMISSIONS)?;
+ let directory_permissions = context.get_property(&definitions::DIRECTORY_PERMISSIONS)?;
Ok(PutFileUnixPermissions {
file_permissions,
@@ -144,11 +139,11 @@ impl PutFileRs {
impl Schedule for PutFileRs {
fn schedule<P: GetProperty, L: Logger>(context: &P, _logger: &L) -> Result<Self, MinifiError> {
let conflict_resolution_strategy =
- context.get_property(&properties::CONFLICT_RESOLUTION)?;
+ context.get_property(&definitions::CONFLICT_RESOLUTION)?;
- let try_make_dirs = context.get_property(&properties::CREATE_DIRS)?;
+ let try_make_dirs = context.get_property(&definitions::CREATE_DIRS)?;
- let maximum_file_count = context.get_property(&properties::MAX_FILE_COUNT)?;
+ let maximum_file_count = context.get_property(&definitions::MAX_FILE_COUNT)?;
let unix_permissions = Self::parse_unix_permissions(context)?;
@@ -202,7 +197,150 @@ impl FlowFileTransform for PutFileRs {
}
}
-pub(crate) mod processor_definition;
-
#[cfg(test)]
-mod tests;
+mod tests {
+ use super::*;
+ use minifi_native::{MockLogger, MockProcessContext};
+
+ #[test]
+ fn schedule_succeeds_with_default_values() {
+ assert!(PutFileRs::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
+ }
+
+ #[test]
+ fn simple_put_file_test() {
+ let mut context = MockProcessContext::new();
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
+ let put_file_dir = temp_dir.path().join("subdir");
+
+ context.properties.insert(
+ "Directory".to_string(),
+ put_file_dir.to_str().unwrap().to_string(),
+ );
+ let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
+
+ let mut input_stream = std::io::Cursor::new("test".as_bytes());
+ context
+ .attributes
+ .insert("filename".to_string(), "test.txt".to_string());
+ let result = put_file
+ .transform(&context, &mut input_stream, &MockLogger::new())
+ .expect("Should succeed");
+
+ assert_eq!(result.target_relationship(), SUCCESS.name);
+
+ let expected_path = temp_dir.path().join("subdir/test.txt");
+ assert!(expected_path.exists());
+ assert_eq!(std::fs::read_to_string(expected_path).unwrap(), "test");
+ }
+
+ #[test]
+ fn put_file_without_create_dirs() {
+ let mut context = MockProcessContext::new();
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
+
+ let put_file_dir = temp_dir.path().join("subdir");
+
+ context.properties.insert(
+ "Directory".to_string(),
+ put_file_dir.to_str().unwrap().to_string(),
+ );
+
+ context.properties.insert(
+ "Create Missing Directories".to_string(),
+ "false".to_string(),
+ );
+
+ let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
+
+ let mut input_stream = std::io::Cursor::new("test".as_bytes());
+ context
+ .attributes
+ .insert("filename".to_string(), "test.txt".to_string());
+ let result = put_file
+ .transform(&context, &mut input_stream, &MockLogger::new())
+ .expect("Should succeed");
+
+ assert_eq!(result.target_relationship(), FAILURE.name);
+
+ let expected_path = temp_dir.path().join("subdir/test.txt");
+ assert!(!expected_path.exists());
+ }
+
+ #[test]
+ fn directory_is_full_counts_only_files() {
+ let mut context = MockProcessContext::new();
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
+
+ context.properties.insert(
+ "Directory".to_string(),
+ temp_dir.path().to_str().unwrap().to_string(),
+ );
+ context
+ .properties
+ .insert("Maximum File Count".to_string(), "2".to_string());
+
+ let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
+
+ let destination = temp_dir.path().join("test.txt");
+
+ // No files yet → not full
+ assert!(!put_file.directory_is_full(&destination));
+
+ // Create a subdirectory; it must not be counted as a file
+ std::fs::create_dir(temp_dir.path().join("subdir")).unwrap();
+ assert!(!put_file.directory_is_full(&destination));
+
+ // Add two files → full
+ std::fs::write(temp_dir.path().join("a.txt"), b"a").unwrap();
+ std::fs::write(temp_dir.path().join("b.txt"), b"b").unwrap();
+ assert!(put_file.directory_is_full(&destination));
+
+ // Remove one file → not full again
+ std::fs::remove_file(temp_dir.path().join("a.txt")).unwrap();
+ assert!(!put_file.directory_is_full(&destination));
+ }
+
+ #[cfg(unix)]
+ #[test]
+ fn put_file_test_permissions() {
+ use std::os::unix::fs::PermissionsExt;
+ let mut context = MockProcessContext::new();
+ let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
+ let put_file_dir = temp_dir.path().join("subdir");
+
+ context.properties.insert(
+ "Directory".to_string(),
+ put_file_dir.to_str().unwrap().to_string(),
+ );
+
+ context
+ .properties
+ .insert("Directory Permissions".to_string(), "0777".to_string());
+
+ context
+ .properties
+ .insert("Permissions".to_string(), "0777".to_string());
+ let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
+
+ let mut input_stream = std::io::Cursor::new("test".as_bytes());
+ context
+ .attributes
+ .insert("filename".to_string(), "test.txt".to_string());
+ let result = put_file
+ .transform(&context, &mut input_stream, &MockLogger::new())
+ .expect("Should succeed");
+
+ assert_eq!(result.target_relationship(), SUCCESS.name);
+
+ let expected_path = temp_dir.path().join("subdir/test.txt");
+ assert!(expected_path.exists());
+ assert_eq!(std::fs::read_to_string(&expected_path).unwrap(), "test");
+ let parent_permissions = std::fs::metadata(put_file_dir).unwrap().permissions();
+ let permissions = expected_path.metadata().unwrap().permissions();
+ // 0o100777: Regular File (10) + No special bits (0) + Full permissions (777)
+ assert_eq!(permissions.mode(), 0o100777);
+ // 0o040777: Directory (04) + No special bits (0) + Full permissions (777)
+ assert_eq!(parent_permissions.mode(), 0o040777);
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/definitions.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/definitions.rs
new file mode 100644
index 000000000..4777d7239
--- /dev/null
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/definitions.rs
@@ -0,0 +1,101 @@
+// 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.
+
+#[cfg(unix)]
+use crate::processors::put_file::unix_permissions::UnixPermission;
+use minifi_native::{
+ NonBlankPath, OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, Property,
+ PropertyDefinition, Relationship, property_definitions,
+};
+
+use super::*;
+
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "Flowfiles that are successfully written to a file are routed to this relationship",
+};
+
+pub(crate) const FAILURE: Relationship = Relationship {
+ name: "failure",
+ description: "Failed files (conflict, write failure, etc.) are transferred to failure",
+};
+
+pub(crate) const DIRECTORY: Property<NonBlankPath> =
+ Property::new("Directory", "The output directory to which to put files")
+ .supports_expression_language()
+ .with_default(".");
+
+pub(crate) const CONFLICT_RESOLUTION: Property<ConflictResolutionStrategy> = Property::new(
+ "Conflict Resolution Strategy",
+ "Indicates what should happen when a file with the same name already exists in the output directory",
+)
+.with_default(ConflictResolutionStrategy::Fail.into_str());
+
+pub(crate) const CREATE_DIRS: Property<bool> = Property::new(
+ "Create Missing Directories",
+ "If true, then missing destination directories will be created. If false, flowfiles are penalized and sent to failure.",
+)
+.with_default("true");
+
+pub(crate) const MAX_FILE_COUNT: Property<Option<u64>> = Property::new(
+ "Maximum File Count",
+ "Specifies the maximum number of files that can exist in the output directory",
+);
+
+#[cfg(unix)]
+pub(crate) const PERMISSIONS: Property<Option<UnixPermission>> = Property::new(
+ "Permissions",
+ "Sets the permissions on the output file to the value of this attribute. Must be an octal number (e.g. 644 or 0755). Not supported on Windows systems.",
+);
+
+#[cfg(unix)]
+pub(crate) const DIRECTORY_PERMISSIONS: Property<Option<UnixPermission>> = Property::new(
+ "Directory Permissions",
+ "Sets the permissions on the directories being created if 'Create Missing Directories' property is set. Must be an octal number (e.g. 644 or 0755). Not supported on Windows systems.",
+);
+
+impl ProcessorDefinition for PutFileRs {
+ const DESCRIPTION: &'static str =
+ "RUST TEST PROCESSOR: Writes the contents of a FlowFile to the local file system.";
+ const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required;
+ const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
+ const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
+ const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
+ const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS, FAILURE];
+
+ #[cfg(unix)]
+ fn properties() -> &'static [PropertyDefinition] {
+ use std::sync::LazyLock;
+ static COMBINED_PROPERTIES: LazyLock<Vec<PropertyDefinition>> = LazyLock::new(|| {
+ let mut props = Vec::new();
+ props.extend_from_slice(property_definitions![
+ DIRECTORY,
+ CONFLICT_RESOLUTION,
+ CREATE_DIRS,
+ MAX_FILE_COUNT,
+ ]);
+ #[cfg(unix)]
+ {
+ props.push(PERMISSIONS.definition());
+ props.push(DIRECTORY_PERMISSIONS.definition());
+ }
+ props
+ });
+
+ &COMBINED_PROPERTIES
+ }
+}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
deleted file mode 100644
index 9b63c3269..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
+++ /dev/null
@@ -1,54 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use minifi_native::{
- OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, property_definitions,
-};
-
-impl ProcessorDefinition for PutFileRs {
- const DESCRIPTION: &'static str =
- "RUST TEST PROCESSOR: Writes the contents of a FlowFile to the local file system.";
- const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required;
- const SUPPORTS_DYNAMIC_PROPERTIES: bool = false;
- const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
- const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
- const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS, FAILURE];
-
- #[cfg(unix)]
- fn properties() -> &'static [PropertyDefinition] {
- use std::sync::LazyLock;
- static COMBINED_PROPERTIES: LazyLock<Vec<PropertyDefinition>> = LazyLock::new(|| {
- let mut props = Vec::new();
- props.extend_from_slice(property_definitions![
- properties::DIRECTORY,
- properties::CONFLICT_RESOLUTION,
- properties::CREATE_DIRS,
- properties::MAX_FILE_COUNT,
- ]);
- #[cfg(unix)]
- {
- props.push(unix_only_properties::PERMISSIONS.definition());
- props.push(unix_only_properties::DIRECTORY_PERMISSIONS.definition());
- }
- props
- });
-
- &COMBINED_PROPERTIES
- }
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/properties.rs
deleted file mode 100644
index 1c1bc8426..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/properties.rs
+++ /dev/null
@@ -1,41 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::ConflictResolutionStrategy;
-use minifi_native::{NonBlankPath, Property};
-
-pub(crate) const DIRECTORY: Property<NonBlankPath> =
- Property::new("Directory", "The output directory to which to put files")
- .supports_expression_language()
- .with_default(".");
-
-pub(crate) const CONFLICT_RESOLUTION: Property<ConflictResolutionStrategy> = Property::new(
- "Conflict Resolution Strategy",
- "Indicates what should happen when a file with the same name already exists in the output directory",
-)
-.with_default(ConflictResolutionStrategy::Fail.into_str());
-
-pub(crate) const CREATE_DIRS: Property<bool> = Property::new(
- "Create Missing Directories",
- "If true, then missing destination directories will be created. If false, flowfiles are penalized and sent to failure.",
-)
-.with_default("true");
-
-pub(crate) const MAX_FILE_COUNT: Property<Option<u64>> = Property::new(
- "Maximum File Count",
- "Specifies the maximum number of files that can exist in the output directory",
-);
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/relationships.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/relationships.rs
deleted file mode 100644
index cd5c506ee..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/relationships.rs
+++ /dev/null
@@ -1,28 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 minifi_native::Relationship;
-
-pub(crate) const SUCCESS: Relationship = Relationship {
- name: "success",
- description: "Flowfiles that are successfully written to a file are routed to this relationship",
-};
-
-pub(crate) const FAILURE: Relationship = Relationship {
- name: "failure",
- description: "Failed files (conflict, write failure, etc.) are transferred to failure",
-};
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/tests.rs
deleted file mode 100644
index d0f2aaa77..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/tests.rs
+++ /dev/null
@@ -1,162 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 super::*;
-use crate::processors::put_file::relationships::{FAILURE, SUCCESS};
-use minifi_native::{MockLogger, MockProcessContext};
-
-#[test]
-fn schedule_succeeds_with_default_values() {
- assert!(PutFileRs::schedule(&MockProcessContext::new(), &MockLogger::new()).is_ok());
-}
-
-#[test]
-fn simple_put_file_test() {
- let mut context = MockProcessContext::new();
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
- let put_file_dir = temp_dir.path().join("subdir");
-
- context.properties.insert(
- "Directory".to_string(),
- put_file_dir.to_str().unwrap().to_string(),
- );
- let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
-
- let mut input_stream = std::io::Cursor::new("test".as_bytes());
- context
- .attributes
- .insert("filename".to_string(), "test.txt".to_string());
- let result = put_file
- .transform(&context, &mut input_stream, &MockLogger::new())
- .expect("Should succeed");
-
- assert_eq!(result.target_relationship(), SUCCESS.name);
-
- let expected_path = temp_dir.path().join("subdir/test.txt");
- assert!(expected_path.exists());
- assert_eq!(std::fs::read_to_string(expected_path).unwrap(), "test");
-}
-
-#[test]
-fn put_file_without_create_dirs() {
- let mut context = MockProcessContext::new();
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
-
- let put_file_dir = temp_dir.path().join("subdir");
-
- context.properties.insert(
- "Directory".to_string(),
- put_file_dir.to_str().unwrap().to_string(),
- );
-
- context.properties.insert(
- "Create Missing Directories".to_string(),
- "false".to_string(),
- );
-
- let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
-
- let mut input_stream = std::io::Cursor::new("test".as_bytes());
- context
- .attributes
- .insert("filename".to_string(), "test.txt".to_string());
- let result = put_file
- .transform(&context, &mut input_stream, &MockLogger::new())
- .expect("Should succeed");
-
- assert_eq!(result.target_relationship(), FAILURE.name);
-
- let expected_path = temp_dir.path().join("subdir/test.txt");
- assert!(!expected_path.exists());
-}
-
-#[test]
-fn directory_is_full_counts_only_files() {
- let mut context = MockProcessContext::new();
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
-
- context.properties.insert(
- "Directory".to_string(),
- temp_dir.path().to_str().unwrap().to_string(),
- );
- context
- .properties
- .insert("Maximum File Count".to_string(), "2".to_string());
-
- let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
-
- let destination = temp_dir.path().join("test.txt");
-
- // No files yet → not full
- assert!(!put_file.directory_is_full(&destination));
-
- // Create a subdirectory; it must not be counted as a file
- std::fs::create_dir(temp_dir.path().join("subdir")).unwrap();
- assert!(!put_file.directory_is_full(&destination));
-
- // Add two files → full
- std::fs::write(temp_dir.path().join("a.txt"), b"a").unwrap();
- std::fs::write(temp_dir.path().join("b.txt"), b"b").unwrap();
- assert!(put_file.directory_is_full(&destination));
-
- // Remove one file → not full again
- std::fs::remove_file(temp_dir.path().join("a.txt")).unwrap();
- assert!(!put_file.directory_is_full(&destination));
-}
-
-#[cfg(unix)]
-#[test]
-fn put_file_test_permissions() {
- use std::os::unix::fs::PermissionsExt;
- let mut context = MockProcessContext::new();
- let temp_dir = tempfile::tempdir().expect("temp dir is required for testing PutFile");
- let put_file_dir = temp_dir.path().join("subdir");
-
- context.properties.insert(
- "Directory".to_string(),
- put_file_dir.to_str().unwrap().to_string(),
- );
-
- context
- .properties
- .insert("Directory Permissions".to_string(), "0777".to_string());
-
- context
- .properties
- .insert("Permissions".to_string(), "0777".to_string());
- let put_file = PutFileRs::schedule(&context, &MockLogger::new()).expect("Should succeed");
-
- let mut input_stream = std::io::Cursor::new("test".as_bytes());
- context
- .attributes
- .insert("filename".to_string(), "test.txt".to_string());
- let result = put_file
- .transform(&context, &mut input_stream, &MockLogger::new())
- .expect("Should succeed");
-
- assert_eq!(result.target_relationship(), SUCCESS.name);
-
- let expected_path = temp_dir.path().join("subdir/test.txt");
- assert!(expected_path.exists());
- assert_eq!(std::fs::read_to_string(&expected_path).unwrap(), "test");
- let parent_permissions = std::fs::metadata(put_file_dir).unwrap().permissions();
- let permissions = expected_path.metadata().unwrap().permissions();
- // 0o100777: Regular File (10) + No special bits (0) + Full permissions (777)
- assert_eq!(permissions.mode(), 0o100777);
- // 0o040777: Directory (04) + No special bits (0) + Full permissions (777)
- assert_eq!(parent_permissions.mode(), 0o040777);
-}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/unix_only_properties.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/unix_only_properties.rs
deleted file mode 100644
index ccaba59ad..000000000
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/unix_only_properties.rs
+++ /dev/null
@@ -1,29 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// 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 crate::processors::put_file::unix_permissions::UnixPermission;
-use minifi_native::Property;
-
-pub(crate) const PERMISSIONS: Property<Option<UnixPermission>> = Property::new(
- "Permissions",
- "Sets the permissions on the output file to the value of this attribute. Must be an octal number (e.g. 644 or 0755). Not supported on Windows systems.",
-);
-
-pub(crate) const DIRECTORY_PERMISSIONS: Property<Option<UnixPermission>> = Property::new(
- "Directory Permissions",
- "Sets the permissions on the directories being created if 'Create Missing Directories' property is set. Must be an octal number (e.g. 644 or 0755). Not supported on Windows systems.",
-);
(nifi-minifi-cpp) 01/02: MINIFICPP-2886 Improve rust api trigger errors
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch minifi_rust_impr_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 358dfef044acacd35a0576f87d089b29f008b75b
Author: Martin Zink <ma...@apache.org>
AuthorDate: Wed Jul 8 10:30:49 2026 +0200
MINIFICPP-2886 Improve rust api trigger errors
---
.../src/processors/asciify_german.rs | 16 +-
.../asciify_german/processor_definition.rs | 4 +-
.../src/processors/asciify_german/tests.rs | 7 +-
.../src/processors/count_actual_logging.rs | 11 +-
.../src/processors/duplicate_text.rs | 11 +-
.../src/processors/generate_flow_file.rs | 8 +-
.../generate_flow_file/processor_definition.rs | 17 +-
.../src/processors/get_file.rs | 8 +-
.../processors/get_file/processor_definition.rs | 27 +--
.../src/processors/get_file/tests.rs | 5 +-
.../src/processors/kamikaze_processor.rs | 14 +-
.../kamikaze_processor/processor_definition.rs | 9 +-
.../src/processors/kamikaze_processor/tests.rs | 7 +-
.../src/processors/log_attribute.rs | 7 +-
.../log_attribute/processor_definition.rs | 21 +-
.../src/processors/lorem_ipsum_cs_user.rs | 9 +-
.../lorem_ipsum_cs_user/processor_definition.rs | 7 +-
.../src/processors/put_file.rs | 7 +-
.../processors/put_file/processor_definition.rs | 36 ++--
.../src/processors/zoo_processor.rs | 14 +-
minifi_rust/minifi_native/src/api/attribute.rs | 11 +-
.../src/api/component_definition_traits.rs | 3 +-
minifi_rust/minifi_native/src/api/errors.rs | 233 +++++++++++++++++++--
.../api/processor_wrappers/complex_processor.rs | 18 +-
.../src/api/processor_wrappers/flow_file_source.rs | 46 ++--
.../flow_file_stream_transform.rs | 74 ++++---
.../api/processor_wrappers/flow_file_transform.rs | 222 ++++++++++++++------
.../src/api/processor_wrappers/utils.rs | 1 +
.../processor_wrappers/utils/with_attributes.rs | 30 +++
minifi_rust/minifi_native/src/api/property.rs | 9 +-
minifi_rust/minifi_native/src/api/raw_processor.rs | 6 +-
.../src/c_ffi/c_ffi_process_session.rs | 23 +-
.../src/c_ffi/c_ffi_processor_definition.rs | 57 ++---
minifi_rust/minifi_native/src/lib.rs | 4 +-
34 files changed, 669 insertions(+), 313 deletions(-)
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs
index a4d8f68ec..6d67c7d05 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs
@@ -20,10 +20,9 @@
use crate::processors::asciify_german::relationships::FAILURE;
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
- FlowFileStreamTransform, GetProperty, InputStream, Logger, MinifiError, OutputStream, Schedule,
- TransformStreamResult,
+ FlowFileStreamTransform, GetProperty, InputStream, Logger, MinifiError, OutputStream,
+ ProcessError, RouteErrorExt, Schedule, TransformStreamResult,
};
-use std::collections::HashMap;
mod relationships;
@@ -46,7 +45,7 @@ impl FlowFileStreamTransform for AsciifyGerman {
input_stream: &mut dyn InputStream,
output_stream: &mut dyn OutputStream,
_logger: &LoggerImpl,
- ) -> Result<TransformStreamResult, MinifiError> {
+ ) -> Result<TransformStreamResult, ProcessError> {
let mut byte = [0u8; 1];
while input_stream.read(&mut byte)? > 0 {
@@ -57,8 +56,8 @@ impl FlowFileStreamTransform for AsciifyGerman {
0xC3 => {
let mut next = [0u8; 1];
if input_stream.read(&mut next)? == 0 {
- // Truncated multi-byte sequence at EOF — treat as malformed input.
- return Ok(TransformStreamResult::route_without_changes(&FAILURE));
+ Err(MinifiError::custom("Truncated multi-byte sequence at EOF"))
+ .route_err_to_failure()?
}
match next[0] {
0xA4 => output_stream.write_all(b"ae")?, // ä
@@ -76,10 +75,7 @@ impl FlowFileStreamTransform for AsciifyGerman {
}
output_stream.flush()?;
- Ok(TransformStreamResult::new(
- &relationships::SUCCESS,
- HashMap::new(),
- ))
+ Ok(TransformStreamResult::new(&relationships::SUCCESS))
}
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs
index 2a64c8a3e..4baa9d9cc 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs
@@ -29,5 +29,7 @@ impl ProcessorDefinition for AsciifyGerman {
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] =
&[super::relationships::SUCCESS, super::relationships::FAILURE];
- const PROPERTIES: &'static [PropertyDefinition] = &[];
+ fn properties() -> &'static [PropertyDefinition] {
+ &[]
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs
index c9b93c658..1a2a77d60 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs
@@ -83,9 +83,6 @@ fn truncated_umlaut_at_eof_routes_to_failure() {
let mut input_stream = BufReader::new(input_bytes);
let mut output_vec: Vec<u8> = Vec::new();
- let result = asciify_german
- .transform(&context, &mut input_stream, &mut output_vec, &logger)
- .expect("Should succeed");
- assert_eq!(result.write_status(), IoState::Cancel);
- assert_eq!(result.target_relationship_name(), FAILURE.name);
+ let result = asciify_german.transform(&context, &mut input_stream, &mut output_vec, &logger);
+ assert!(result.is_err());
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs
index 518751d62..be37674ba 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs
@@ -20,8 +20,8 @@
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
GetProperty, Logger, MinifiError, MutTrigger, OnTriggerResult, OutputAttribute, ProcessContext,
- ProcessSession, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition,
- Relationship, Schedule, debug, info, trace,
+ ProcessError, ProcessSession, ProcessorDefinition, ProcessorInputRequirement,
+ PropertyDefinition, Relationship, Schedule, debug, info, trace,
};
#[derive(Debug, ComponentIdentifier)]
@@ -51,7 +51,7 @@ impl MutTrigger for CountActualLogging {
_context: &mut PC,
_session: &mut PS,
logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -72,5 +72,8 @@ impl ProcessorDefinition for CountActualLogging {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[];
- const PROPERTIES: &'static [PropertyDefinition] = &[];
+
+ fn properties() -> &'static [PropertyDefinition] {
+ &[]
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs
index ee4ae055e..ee049cd4f 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs
@@ -18,10 +18,9 @@
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
GetAttribute, GetControllerService, GetProperty, InputStream, Logger, MinifiError,
- MutFlowFileStreamTransform, OutputAttribute, OutputStream, ProcessorDefinition,
+ MutFlowFileStreamTransform, OutputAttribute, OutputStream, ProcessError, ProcessorDefinition,
ProcessorInputRequirement, PropertyDefinition, Relationship, Schedule, TransformStreamResult,
};
-use std::collections::HashMap;
#[derive(Debug, ComponentIdentifier)]
pub(crate) struct DuplicateStreamText {}
@@ -50,13 +49,13 @@ impl MutFlowFileStreamTransform for DuplicateStreamText {
input_stream: &mut dyn InputStream,
output_stream: &mut dyn OutputStream,
_logger: &LoggerImpl,
- ) -> Result<TransformStreamResult, MinifiError> {
+ ) -> Result<TransformStreamResult, ProcessError> {
let mut byte = [0u8; 1];
while input_stream.read(&mut byte)? > 0 {
let _ = output_stream.write(&byte)?;
let _ = output_stream.write(&byte)?;
}
- Ok(TransformStreamResult::new(&SUCCESS, HashMap::new()))
+ Ok(TransformStreamResult::new(&SUCCESS))
}
}
@@ -67,5 +66,7 @@ impl ProcessorDefinition for DuplicateStreamText {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = &[];
+ fn properties() -> &'static [PropertyDefinition] {
+ &[]
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
index 959ef7820..b5c9045e5 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
@@ -19,8 +19,8 @@
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
- GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule,
- Trigger,
+ GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, Schedule, Trigger,
};
use rand::RngExt;
use rand::distr::Alphanumeric;
@@ -147,7 +147,7 @@ impl Trigger for GenerateFlowFileRs {
context: &mut PC,
session: &mut PS,
_logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -164,7 +164,7 @@ impl Trigger for GenerateFlowFileRs {
context
.get_raw_property(&properties::CUSTOM_TEXT, None)?
.ok_or_else(|| {
- MinifiError::trigger_err(
+ MinifiError::custom(
"GenerateFlowFile is in CustomText mode but the \"Custom Text\" \
property is not set at trigger time",
)
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
index d3f9a9391..a3e0e8a68 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
@@ -29,11 +29,14 @@ impl ProcessorDefinition for GenerateFlowFileRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- FILE_SIZE,
- BATCH_SIZE,
- DATA_FORMAT,
- UNIQUE_FLOW_FILES,
- CUSTOM_TEXT
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ FILE_SIZE,
+ BATCH_SIZE,
+ DATA_FORMAT,
+ UNIQUE_FLOW_FILES,
+ CUSTOM_TEXT
+ ];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
index ab2f0c037..7ec9f4130 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
@@ -26,8 +26,8 @@ use crate::processors::get_file::properties::{
};
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
- GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession,
- Schedule, Trigger, debug, info, trace, warn,
+ GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, Schedule, Trigger, debug, info, trace, warn,
};
use std::collections::VecDeque;
use std::error;
@@ -225,7 +225,7 @@ impl Schedule for GetFileRs {
{
let input_directory = context.get_property(&DIRECTORY)?;
if !input_directory.is_dir() {
- return Err(MinifiError::schedule_err(format!(
+ return Err(MinifiError::custom(format!(
"{:?} is not a valid directory",
input_directory
)));
@@ -269,7 +269,7 @@ impl Trigger for GetFileRs {
context: &mut PC,
session: &mut PS,
logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
index 59171d3d0..13dbbbbdc 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
@@ -33,16 +33,19 @@ impl ProcessorDefinition for GetFileRs {
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] =
&[ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- DIRECTORY,
- POLLING_INTERVAL,
- RECURSE,
- KEEP_SOURCE_FILE,
- MIN_AGE,
- MAX_AGE,
- MIN_SIZE,
- MAX_SIZE,
- IGNORE_HIDDEN_FILES,
- BATCH_SIZE,
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ DIRECTORY,
+ POLLING_INTERVAL,
+ RECURSE,
+ KEEP_SOURCE_FILE,
+ MIN_AGE,
+ MAX_AGE,
+ MIN_SIZE,
+ MAX_SIZE,
+ IGNORE_HIDDEN_FILES,
+ BATCH_SIZE,
+ ];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
index 50691c773..d7b9ed96f 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
@@ -38,10 +38,7 @@ fn schedule_fails_with_invalid_input_dir() {
"Input Directory".to_string(),
"/invalid_directory".to_string(),
);
- assert!(matches!(
- GetFileRs::schedule(&context, &MockLogger::new()),
- Err(MinifiError::ScheduleError(_))
- ));
+ assert!(GetFileRs::schedule(&context, &MockLogger::new()).is_err());
}
#[test]
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
index f6f843576..0faa04f18 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
@@ -25,8 +25,8 @@ use crate::processors::kamikaze_processor::properties::{
};
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
- GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule,
- Trigger,
+ GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, Schedule, Trigger,
};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -57,7 +57,7 @@ impl Schedule for KamikazeProcessorRs {
let schedule_behaviour = context.get_property(&SCHEDULE_BEHAVIOUR)?;
match schedule_behaviour {
- KamikazeBehaviour::ReturnErr => Err(MinifiError::schedule_err(
+ KamikazeBehaviour::ReturnErr => Err(MinifiError::custom(
"it was designed to fail during schedule",
)),
KamikazeBehaviour::ReturnOk => Ok(KamikazeProcessorRs { trigger_behaviour }),
@@ -81,16 +81,16 @@ impl Trigger for KamikazeProcessorRs {
context: &mut PC,
_session: &mut PS,
_logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
L: Logger,
{
match self.trigger_behaviour {
- KamikazeBehaviour::ReturnErr => Err(MinifiError::trigger_err(
- "it was designed to fail in trigger",
- )),
+ KamikazeBehaviour::ReturnErr => {
+ Err(MinifiError::custom("it was designed to fail in trigger").into())
+ }
KamikazeBehaviour::ReturnOk => Ok(OnTriggerResult::Ok),
KamikazeBehaviour::Panic => {
panic!("KamikazeProcessor::trigger panic")
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
index 51c8ff0c8..f9a47d115 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
@@ -28,8 +28,9 @@ impl ProcessorDefinition for KamikazeProcessorRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- properties::SCHEDULE_BEHAVIOUR,
- properties::TRIGGER_BEHAVIOUR
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
index dd6e4794f..2b7f52166 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
@@ -17,8 +17,7 @@
use super::*;
use crate::processors::kamikaze_processor::properties::{SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR};
-use minifi_native::MinifiError::{ScheduleError, TriggerError};
-use minifi_native::{MockLogger, MockProcessContext, MockProcessSession};
+use minifi_native::{MockLogger, MockProcessContext, MockProcessSession, ProcessError};
use std::panic::AssertUnwindSafe;
#[test]
@@ -36,7 +35,7 @@ fn on_schedule_err() {
"ReturnErr".to_string(),
);
let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new());
- assert!(matches!(processor, Err(ScheduleError(_))));
+ assert!(matches!(processor, Err(MinifiError::CustomError(_))));
}
#[test]
@@ -78,7 +77,7 @@ fn on_trigger_err() {
let mut session = MockProcessSession::new();
assert!(matches!(
processor.trigger(&mut context, &mut session, &MockLogger::new()),
- Err(TriggerError(_))
+ Err(ProcessError::Fatal(MinifiError::CustomError(_)))
));
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
index 48de81360..329e281f4 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
@@ -21,8 +21,9 @@ use crate::processors::log_attribute::properties::{FLOW_FILES_TO_LOG, LOG_LEVEL,
use minifi_native::StandardPropertyValidator::NonBlankValidator;
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
- GetProperty, LogLevel, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession,
- PropertyConstraints, PropertySchema, PropertyType, Schedule, Trigger, debug, log, trace,
+ GetProperty, LogLevel, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, PropertyConstraints, PropertySchema, PropertyType, Schedule, Trigger, debug,
+ log, trace,
};
mod properties;
@@ -104,7 +105,7 @@ impl Trigger for LogAttributeRs {
_context: &mut PC,
session: &mut PS,
logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
index b5732645d..91e0b3fed 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
@@ -30,13 +30,16 @@ impl ProcessorDefinition for LogAttributeRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- LOG_LEVEL,
- ATTRIBUTES_TO_LOG,
- ATTRIBUTES_TO_IGNORE,
- LOG_PAYLOAD,
- LOG_PREFIX,
- FLOW_FILES_TO_LOG,
- HEX_ENCODE_PAYLOAD,
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ LOG_LEVEL,
+ ATTRIBUTES_TO_LOG,
+ ATTRIBUTES_TO_IGNORE,
+ LOG_PAYLOAD,
+ LOG_PREFIX,
+ FLOW_FILES_TO_LOG,
+ HEX_ENCODE_PAYLOAD,
+ ];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
index f8a684d9f..4d32739e2 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
@@ -25,9 +25,8 @@ use crate::processors::lorem_ipsum_cs_user::relationships::SUCCESS;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
Content, FlowFileSource, GeneratedFlowFile, GetControllerService, GetProperty, Logger,
- MinifiError, Schedule, trace,
+ MinifiError, ProcessError, Schedule, trace,
};
-use std::collections::HashMap;
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
#[derive(
@@ -59,7 +58,7 @@ impl FlowFileSource for LoremIpsumCSUser {
&self,
context: &'a mut Context,
logger: &LoggerImpl,
- ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError> {
+ ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError> {
trace!(logger, "generate call {:?}", self);
let dummy_controller_service = context.get_controller_service(&DUMMY_CONTROLLER_SERVICE)?;
trace!(
@@ -72,15 +71,13 @@ impl FlowFileSource for LoremIpsumCSUser {
let generated_flow_file = GeneratedFlowFile::new(
&SUCCESS,
Some(Content::from(controller_service.data.clone())),
- HashMap::new(),
);
Ok(vec![generated_flow_file])
}
WriteMethod::Stream => {
let reader = controller_service.data.as_bytes();
let content = Content::Stream(Box::new(reader));
- let generated_flow_file =
- GeneratedFlowFile::new(&SUCCESS, Some(content), HashMap::new());
+ let generated_flow_file = GeneratedFlowFile::new(&SUCCESS, Some(content));
Ok(vec![generated_flow_file])
}
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
index 78c09ef0c..a80605035 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
@@ -31,6 +31,9 @@ impl ProcessorDefinition for LoremIpsumCSUser {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] =
- property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
index e81c861ca..1dd63ec2d 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
@@ -22,7 +22,7 @@ use crate::processors::put_file::unix_permissions::PutFileUnixPermissions;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
FlowFileTransform, GetAttribute, GetControllerService, GetId, GetProperty, InputStream, Logger,
- MinifiError, Schedule, TransformedFlowFile, trace, unwrap_or_route, warn,
+ MinifiError, ProcessError, RouteErrorExt, Schedule, TransformedFlowFile, trace, warn,
};
use std::path::{Path, PathBuf};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -171,11 +171,10 @@ impl FlowFileTransform for PutFileRs {
context: &Context,
input_stream: &'a mut dyn InputStream,
logger: &LoggerImpl,
- ) -> Result<TransformedFlowFile<'a>, MinifiError> {
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
trace!(logger, "on_trigger: {:?}", self);
- let destination_path =
- unwrap_or_route!(Self::get_destination_path(context), &FAILURE, logger);
+ let destination_path = Self::get_destination_path(context).route_err_to_failure()?;
if self.directory_is_full(&destination_path) {
warn!(logger, "Directory is full");
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
index 93be840a4..9b63c3269 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
@@ -30,21 +30,25 @@ impl ProcessorDefinition for PutFileRs {
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS, FAILURE];
- #[cfg(windows)]
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- properties::DIRECTORY,
- properties::CONFLICT_RESOLUTION,
- properties::CREATE_DIRS,
- properties::MAX_FILE_COUNT,
- ];
-
#[cfg(unix)]
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- properties::DIRECTORY,
- properties::CONFLICT_RESOLUTION,
- properties::CREATE_DIRS,
- properties::MAX_FILE_COUNT,
- unix_only_properties::PERMISSIONS,
- unix_only_properties::DIRECTORY_PERMISSIONS,
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ use std::sync::LazyLock;
+ static COMBINED_PROPERTIES: LazyLock<Vec<PropertyDefinition>> = LazyLock::new(|| {
+ let mut props = Vec::new();
+ props.extend_from_slice(property_definitions![
+ properties::DIRECTORY,
+ properties::CONFLICT_RESOLUTION,
+ properties::CREATE_DIRS,
+ properties::MAX_FILE_COUNT,
+ ]);
+ #[cfg(unix)]
+ {
+ props.push(unix_only_properties::PERMISSIONS.definition());
+ props.push(unix_only_properties::DIRECTORY_PERMISSIONS.definition());
+ }
+ props
+ });
+
+ &COMBINED_PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
index 86d67c779..6eabf7a4d 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
@@ -21,8 +21,8 @@ use crate::controller_services::animal_controller_apis::{
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
GetProperty, Logger, MinifiError, OnTriggerResult, OutputAttribute, ProcessContext,
- ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property, PropertyDefinition,
- Relationship, Schedule, Trigger, critical, info, property_definitions,
+ ProcessError, ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property,
+ PropertyDefinition, Relationship, Schedule, Trigger, critical, info, property_definitions,
};
pub(crate) const CAN_FLY_SERVICE: Property<dyn CanFlyControllerApi> =
@@ -52,7 +52,7 @@ impl Trigger for ZooProcessorRs {
context: &mut Context,
_session: &mut Session,
logger: &Lggr,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
Context: ProcessContext,
Session: ProcessSession<FlowFile = Context::FlowFile>,
@@ -81,6 +81,10 @@ impl ProcessorDefinition for ZooProcessorRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[];
- const PROPERTIES: &'static [PropertyDefinition] =
- property_definitions![CAN_FLY_SERVICE, NUMBER_OF_LEGS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![CAN_FLY_SERVICE, NUMBER_OF_LEGS];
+
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/minifi_native/src/api/attribute.rs b/minifi_rust/minifi_native/src/api/attribute.rs
index 571ea7ce7..ed0301654 100644
--- a/minifi_rust/minifi_native/src/api/attribute.rs
+++ b/minifi_rust/minifi_native/src/api/attribute.rs
@@ -16,6 +16,7 @@
// under the License.
use crate::MinifiError;
+use std::borrow::Cow;
pub struct OutputAttribute {
pub name: &'static str,
@@ -23,10 +24,18 @@ pub struct OutputAttribute {
pub description: &'static str,
}
+impl From<&OutputAttribute> for Cow<'static, str> {
+ fn from(attr: &OutputAttribute) -> Self {
+ Cow::Borrowed(attr.name)
+ }
+}
+
pub trait GetAttribute {
fn get_attribute(&self, name: &str) -> Result<Option<String>, MinifiError>;
fn get_required_attribute(&self, name: &str) -> Result<String, MinifiError> {
self.get_attribute(name)?
- .ok_or(MinifiError::missing_required_attribute(name.to_owned()))
+ .ok_or(MinifiError::MissingRequiredAttribute(
+ name.to_owned().into(),
+ ))
}
}
diff --git a/minifi_rust/minifi_native/src/api/component_definition_traits.rs b/minifi_rust/minifi_native/src/api/component_definition_traits.rs
index 5e0e8a126..c1f473f42 100644
--- a/minifi_rust/minifi_native/src/api/component_definition_traits.rs
+++ b/minifi_rust/minifi_native/src/api/component_definition_traits.rs
@@ -31,7 +31,8 @@ pub trait ProcessorDefinition {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute];
const RELATIONSHIPS: &'static [Relationship];
- const PROPERTIES: &'static [PropertyDefinition];
+ /// fn to allow concatenation for complex processors
+ fn properties() -> &'static [PropertyDefinition];
}
pub trait ControllerServiceDefinition: Sized + 'static {
diff --git a/minifi_rust/minifi_native/src/api/errors.rs b/minifi_rust/minifi_native/src/api/errors.rs
index 2c833a997..96ca1156b 100644
--- a/minifi_rust/minifi_native/src/api/errors.rs
+++ b/minifi_rust/minifi_native/src/api/errors.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use minifi_native::{LogLevel, Relationship};
use minifi_native_sys::minifi_status;
use std::borrow::Cow;
use std::error::Error;
@@ -35,19 +36,150 @@ pub enum ParseError {
Other,
}
+#[derive(Debug)]
+pub struct RouteError {
+ pub relationship: Cow<'static, str>,
+ pub source: Box<dyn Error + Send + Sync + 'static>,
+ pub log_level: LogLevel,
+}
+
+impl RouteError {
+ pub(crate) fn log<L: crate::Logger>(&self, logger: &L) {
+ logger.log(
+ self.log_level,
+ format_args!(
+ "Routing flow file to '{}': {}",
+ self.relationship, self.source
+ ),
+ );
+ }
+}
+
+impl fmt::Display for RouteError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(
+ f,
+ "route to '{}' due to: {}",
+ self.relationship, self.source
+ )
+ }
+}
+
+impl Error for RouteError {}
+
+/// The top-level error returned by a processor trigger. It is exactly one of
+/// two kinds:
+///
+/// * [`ProcessError::Route`] — transfer the current flow file to a relationship
+/// and commit (no rollback).
+/// * [`ProcessError::Fatal`] — a real error that must fail the trigger so the
+/// agent rolls back the session.
+#[derive(Debug)]
+pub enum ProcessError {
+ Route(RouteError),
+ Fatal(MinifiError),
+}
+
+impl From<RouteError> for ProcessError {
+ fn from(err: RouteError) -> Self {
+ ProcessError::Route(err)
+ }
+}
+
+impl From<MinifiError> for ProcessError {
+ fn from(err: MinifiError) -> Self {
+ ProcessError::Fatal(err)
+ }
+}
+
+macro_rules! process_error_from_fatal {
+ ($($t:ty),* $(,)?) => {
+ $(
+ impl From<$t> for ProcessError {
+ fn from(err: $t) -> Self {
+ ProcessError::Fatal(MinifiError::from(err))
+ }
+ }
+ )*
+ };
+}
+
+process_error_from_fatal!(
+ std::io::Error,
+ strum::ParseError,
+ ParseBoolError,
+ ParseIntError,
+ humantime::DurationError,
+ byte_unit::ParseError,
+ NulError,
+ ParseFloatError,
+ std::convert::Infallible,
+);
+
+impl fmt::Display for ProcessError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ ProcessError::Route(err) => write!(f, "{}", err),
+ ProcessError::Fatal(err) => write!(f, "{}", err),
+ }
+ }
+}
+
+impl Error for ProcessError {}
+
+pub trait RouteErrorExt<T> {
+ fn route_err(self, rel: &Relationship, level: LogLevel) -> Result<T, ProcessError>;
+
+ fn route_to<S: Into<Cow<'static, str>>>(
+ self,
+ relationship: S,
+ level: LogLevel,
+ ) -> Result<T, ProcessError>;
+
+ fn route_err_to_failure(self) -> Result<T, ProcessError>;
+}
+
+impl<T, E> RouteErrorExt<T> for Result<T, E>
+where
+ E: Into<Box<dyn Error + Send + Sync + 'static>>,
+{
+ fn route_err(self, rel: &Relationship, level: LogLevel) -> Result<T, ProcessError> {
+ self.route_to(rel.name, level)
+ }
+
+ fn route_to<S: Into<Cow<'static, str>>>(
+ self,
+ relationship: S,
+ level: LogLevel,
+ ) -> Result<T, ProcessError> {
+ self.map_err(|e| {
+ ProcessError::Route(RouteError {
+ relationship: relationship.into(),
+ source: e.into(),
+ log_level: level,
+ })
+ })
+ }
+
+ fn route_err_to_failure(self) -> Result<T, ProcessError> {
+ self.route_to("failure", LogLevel::Warn)
+ }
+}
+
#[derive(Debug)]
pub enum MinifiError {
UnknownError,
StatusError((Cow<'static, str>, NonZeroU32)),
MissingRequiredAttribute(Cow<'static, str>),
MissingRequiredProperty(Cow<'static, str>),
- ControllerServiceError(Cow<'static, str>),
+ UnscheduledProcessor,
ValidationError(Cow<'static, str>),
- ScheduleError(Cow<'static, str>),
- TriggerError(Cow<'static, str>),
+ CustomError(Cow<'static, str>),
Parse(ParseError),
MissingFlowFileError,
IoError(std::io::Error),
+
+ Other(Box<dyn Error + Send + Sync + 'static>),
}
impl From<std::io::Error> for MinifiError {
@@ -124,33 +256,24 @@ impl MinifiError {
}
}
- pub fn validation_err<S: Into<Cow<'static, str>>>(msg: S) -> Self {
+ pub fn validation<S: Into<Cow<'static, str>>>(msg: S) -> Self {
MinifiError::ValidationError(msg.into())
}
- pub fn schedule_err<S: Into<Cow<'static, str>>>(msg: S) -> Self {
- MinifiError::ScheduleError(msg.into())
- }
-
- pub fn trigger_err<S: Into<Cow<'static, str>>>(msg: S) -> Self {
- MinifiError::TriggerError(msg.into())
- }
-
- pub fn missing_required_property<S: Into<Cow<'static, str>>>(msg: S) -> Self {
- MinifiError::MissingRequiredProperty(msg.into())
- }
-
- pub fn missing_required_attribute<S: Into<Cow<'static, str>>>(msg: S) -> Self {
- MinifiError::MissingRequiredAttribute(msg.into())
- }
-
- pub fn controller_service_err<S: Into<Cow<'static, str>>>(msg: S) -> Self {
- MinifiError::ControllerServiceError(msg.into())
+ pub fn custom<S: Into<Cow<'static, str>>>(msg: S) -> Self {
+ MinifiError::CustomError(msg.into())
}
pub fn parse_err() -> Self {
MinifiError::Parse(ParseError::Other)
}
+
+ pub fn other<E>(err: E) -> Self
+ where
+ E: Into<Box<dyn Error + Send + Sync + 'static>>,
+ {
+ MinifiError::Other(err.into())
+ }
}
impl fmt::Display for MinifiError {
@@ -177,9 +300,75 @@ impl fmt::Display for MinifiError {
}
_ => write!(f, "{} (Unknown Status Code: {})", context, code),
},
+ MinifiError::Other(err) => write!(f, "Custom error: {}", err),
_ => write!(f, "{:?}", self),
}
}
}
impl Error for MinifiError {}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn io_err() -> std::io::Error {
+ std::io::Error::other("boom")
+ }
+
+ #[test]
+ fn route_err_to_failure_uses_warn() {
+ let res: Result<(), std::io::Error> = Err(io_err());
+ match res.route_err_to_failure() {
+ Err(ProcessError::Route(route)) => {
+ assert_eq!(route.relationship.as_ref(), "failure");
+ assert_eq!(route.log_level, LogLevel::Warn);
+ assert_eq!(route.source.to_string(), "boom");
+ }
+ other => panic!("expected a route error, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn route_err_uses_the_relationships_name() {
+ const REJECT: Relationship = Relationship {
+ name: "reject",
+ description: "",
+ };
+ let res: Result<(), std::io::Error> = Err(io_err());
+ match res.route_err(&REJECT, LogLevel::Info) {
+ Err(ProcessError::Route(route)) => {
+ assert_eq!(route.relationship.as_ref(), "reject");
+ assert_eq!(route.log_level, LogLevel::Info);
+ }
+ other => panic!("expected a route error, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn ok_values_pass_through_unchanged() {
+ let res: Result<u8, std::io::Error> = Ok(5);
+ assert_eq!(res.route_err_to_failure().unwrap(), 5);
+ }
+
+ #[test]
+ fn minifi_error_converts_to_fatal_via_from() {
+ let pe: ProcessError = MinifiError::custom("nope").into();
+ assert!(matches!(
+ pe,
+ ProcessError::Fatal(MinifiError::CustomError(_))
+ ));
+ }
+
+ #[test]
+ fn raw_error_question_mark_becomes_fatal() {
+ fn inner() -> Result<(), ProcessError> {
+ Err(io_err())?;
+ Ok(())
+ }
+ assert!(matches!(
+ inner(),
+ Err(ProcessError::Fatal(MinifiError::IoError(_)))
+ ));
+ }
+}
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs
index 24f29384f..7699f4b33 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs
@@ -18,7 +18,7 @@
use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger};
use crate::{
ComponentIdentifier, Logger, MinifiError, MultiThreaded, OnTriggerResult, ProcessContext,
- ProcessSession, Processor, ProcessorDefinition, Schedule, SingleThreaded,
+ ProcessError, ProcessSession, Processor, ProcessorDefinition, Schedule, SingleThreaded,
};
pub trait MutTrigger {
@@ -27,7 +27,7 @@ pub trait MutTrigger {
context: &mut Ctx,
session: &mut Session,
logger: &Lggr,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
Ctx: ProcessContext,
Session: ProcessSession<FlowFile = Ctx::FlowFile>,
@@ -40,7 +40,7 @@ pub trait Trigger {
context: &mut Context,
session: &mut Session,
logger: &Lggr,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
Context: ProcessContext,
Session: ProcessSession<FlowFile = Context::FlowFile>,
@@ -59,7 +59,7 @@ where
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -67,9 +67,7 @@ where
if let Some(ref mut scheduled_impl) = self.scheduled_impl {
scheduled_impl.trigger(context, session, &self.logger)
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
@@ -84,7 +82,7 @@ where
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -92,9 +90,7 @@ where
if let Some(ref scheduled_impl) = self.scheduled_impl {
scheduled_impl.trigger(context, session, &self.logger)
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs
index 1f049ff12..08fd8fbf4 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs
@@ -17,42 +17,42 @@
use crate::api::processor_wrappers::utils::flow_file_content::Content;
use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger};
+use crate::{FlowFileAttribute, impl_with_attributes};
use crate::{
GetControllerService, GetProperty, Logger, MinifiError, MultiThreaded, OnTriggerResult,
- ProcessContext, ProcessSession, Processor, Relationship, Schedule, SingleThreaded,
+ ProcessContext, ProcessError, ProcessSession, Processor, Relationship, Schedule,
+ SingleThreaded,
};
-use std::collections::HashMap;
+use std::borrow::Cow;
pub struct GeneratedFlowFile<'a> {
- target_relationship_name: &'static str,
+ target_relationship_name: Cow<'static, str>,
new_content: Option<Content<'a>>,
- attributes_to_add: HashMap<String, String>,
+ attributes_to_add: Vec<FlowFileAttribute>,
}
impl<'a> GeneratedFlowFile<'a> {
- pub fn new(
- target_relationship: &'a Relationship,
- new_content: Option<Content<'a>>,
- attributes_to_add: HashMap<String, String>,
- ) -> Self {
+ pub fn new(target_relationship: &'a Relationship, new_content: Option<Content<'a>>) -> Self {
Self {
- target_relationship_name: target_relationship.name,
+ target_relationship_name: Cow::Borrowed(target_relationship.name),
new_content,
- attributes_to_add,
+ attributes_to_add: Vec::new(),
}
}
- pub fn target_relationship_name(&self) -> &'static str {
- self.target_relationship_name
+ pub fn target_relationship_name(&self) -> &str {
+ &self.target_relationship_name
}
}
+impl_with_attributes!(GeneratedFlowFile<'a>);
+
pub trait FlowFileSource {
fn generate<'a, Context: GetProperty + GetControllerService, LoggerImpl: Logger>(
&self,
context: &'a mut Context,
logger: &LoggerImpl,
- ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError>;
+ ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError>;
}
pub trait MutFlowFileSource {
@@ -60,13 +60,13 @@ pub trait MutFlowFileSource {
&mut self,
context: &'a mut Context,
logger: &LoggerImpl,
- ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError>;
+ ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError>;
}
fn handle_generated_flow_files<PC, PS>(
session: &mut PS,
generated_flow_files: Vec<GeneratedFlowFile>,
-) -> Result<OnTriggerResult, MinifiError>
+) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -85,7 +85,7 @@ where
for (k, v) in &new_flow_file_data.attributes_to_add {
session.set_attribute(&mut ff, k, v)?;
}
- session.transfer(ff, new_flow_file_data.target_relationship_name)?;
+ session.transfer(ff, new_flow_file_data.target_relationship_name.as_ref())?;
}
Ok(OnTriggerResult::Ok)
}
@@ -102,7 +102,7 @@ where
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -111,9 +111,7 @@ where
let files = scheduled_impl.generate(context, &self.logger)?;
handle_generated_flow_files::<PC, PS>(session, files)
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
@@ -128,7 +126,7 @@ where
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -137,9 +135,7 @@ where
let files = scheduled_impl.generate(context, &self.logger)?;
handle_generated_flow_files::<PC, PS>(session, files)
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs
index ece4244b9..5b6a2643a 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs
@@ -18,45 +18,50 @@
use crate::api::process_session::IoState;
use crate::api::processor_wrappers::utils::context_session_flowfile_bundle::ContextSessionFlowFileBundle;
use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger};
+use crate::{FlowFileAttribute, impl_with_attributes};
use crate::{
GetAttribute, GetControllerService, GetProperty, InputStream, LogLevel, Logger, MinifiError,
- MultiThreaded, OnTriggerResult, OutputStream, ProcessContext, ProcessSession, Processor,
- Relationship, Schedule, SingleThreaded,
+ MultiThreaded, OnTriggerResult, OutputStream, ProcessContext, ProcessError, ProcessSession,
+ Processor, Relationship, Schedule, SingleThreaded,
};
-use std::collections::HashMap;
+use std::borrow::Cow;
pub struct TransformStreamResult {
- target_relationship_name: &'static str,
- attributes_to_add: HashMap<String, String>,
+ target_relationship_name: Cow<'static, str>,
+ attributes_to_add: Vec<FlowFileAttribute>,
write_status: IoState,
}
impl TransformStreamResult {
- pub fn new(
- target_relationship: &Relationship,
- attributes_to_add: HashMap<String, String>,
- ) -> Self {
+ pub fn new(target_relationship: &Relationship) -> Self {
Self {
- target_relationship_name: target_relationship.name,
- attributes_to_add,
+ target_relationship_name: Cow::Borrowed(target_relationship.name),
+ attributes_to_add: Vec::new(),
write_status: IoState::Ok,
}
}
pub fn route_without_changes(target_relationship: &Relationship) -> Self {
+ Self::route_without_changes_by_name(Cow::Borrowed(target_relationship.name))
+ }
+
+ pub fn route_without_changes_by_name(relationship: Cow<'static, str>) -> Self {
Self {
- target_relationship_name: target_relationship.name,
- attributes_to_add: HashMap::new(),
+ target_relationship_name: relationship,
+ attributes_to_add: Vec::new(),
write_status: IoState::Cancel,
}
}
- pub fn target_relationship_name(&self) -> &'static str {
- self.target_relationship_name
+ pub fn target_relationship_name(&self) -> &str {
+ &self.target_relationship_name
}
- pub fn get_attribute(&self, name: &str) -> Option<String> {
- self.attributes_to_add.get(name).cloned()
+ pub fn get_attribute(&self, name: &str) -> Option<&str> {
+ self.attributes_to_add
+ .iter()
+ .find(|(k, _)| k == name)
+ .map(|(_, v)| v.as_ref())
}
pub fn write_status(&self) -> IoState {
@@ -64,6 +69,8 @@ impl TransformStreamResult {
}
}
+impl_with_attributes!(TransformStreamResult);
+
pub trait FlowFileStreamTransform {
fn transform<Ctx: GetProperty + GetControllerService + GetAttribute, LoggerImpl: Logger>(
&self,
@@ -71,7 +78,7 @@ pub trait FlowFileStreamTransform {
input_stream: &mut dyn InputStream,
output_stream: &mut dyn OutputStream,
logger: &LoggerImpl,
- ) -> Result<TransformStreamResult, MinifiError>;
+ ) -> Result<TransformStreamResult, ProcessError>;
}
pub trait MutFlowFileStreamTransform {
@@ -81,7 +88,7 @@ pub trait MutFlowFileStreamTransform {
input_stream: &mut dyn InputStream,
output_stream: &mut dyn OutputStream,
logger: &LoggerImpl,
- ) -> Result<TransformStreamResult, MinifiError>;
+ ) -> Result<TransformStreamResult, ProcessError>;
}
pub struct FlowFileStreamTransformProcessorType {}
@@ -91,7 +98,7 @@ fn handle_stream_transform<PC, PS, L, F>(
session: &mut PS,
logger: &L,
mut transform_fn: F,
-) -> Result<OnTriggerResult, MinifiError>
+) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -100,14 +107,23 @@ where
&ContextSessionFlowFileBundle<PC, PS>,
&mut dyn InputStream,
&mut dyn OutputStream,
- ) -> Result<TransformStreamResult, MinifiError>,
+ ) -> Result<TransformStreamResult, ProcessError>,
{
if let Some(mut flow_file) = session.get() {
let simple_context = ContextSessionFlowFileBundle::new(context, session, Some(&flow_file));
let (relationship, attrs) = session.read_stream(&flow_file, |input_stream| {
session.write_stream(&flow_file, |output_stream| {
- let transformed = transform_fn(&simple_context, input_stream, output_stream)?;
+ let transformed = match transform_fn(&simple_context, input_stream, output_stream) {
+ Ok(t) => t,
+ Err(ProcessError::Route(route)) => {
+ route.log(logger);
+ TransformStreamResult::route_without_changes_by_name(route.relationship)
+ }
+ Err(ProcessError::Fatal(e)) => {
+ return Err(e);
+ }
+ };
Ok((
(
@@ -123,7 +139,7 @@ where
session.set_attribute(&mut flow_file, &k, &v)?;
}
- session.transfer(flow_file, relationship)?;
+ session.transfer(flow_file, relationship.as_ref())?;
Ok(OnTriggerResult::Ok)
} else {
@@ -142,7 +158,7 @@ where
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -152,9 +168,7 @@ where
scheduled_impl.transform(ctx, input, output, &self.logger)
})
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
@@ -169,7 +183,7 @@ where
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -179,9 +193,7 @@ where
scheduled_impl.transform(ctx, input, output, &self.logger)
})
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs
index ef2342b5b..462b70973 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use crate::api::InputStream;
use crate::api::flow_file::GetId;
use crate::api::processor::Processor;
use crate::api::processor_wrappers::utils::context_session_flowfile_bundle::ContextSessionFlowFileBundle;
@@ -24,50 +23,68 @@ use crate::api::property::{GetControllerService, GetProperty};
use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger};
use crate::{
GetAttribute, LogLevel, Logger, MinifiError, MultiThreaded, OnTriggerResult, ProcessContext,
- ProcessSession, Relationship, Schedule, SingleThreaded, info,
+ ProcessError, ProcessSession, Relationship, Schedule, SingleThreaded, impl_with_attributes,
+ info,
};
-use std::collections::HashMap;
+
+use minifi_native::InputStream;
+use std::borrow::Cow;
+
+pub type FlowFileAttribute = (Cow<'static, str>, Cow<'static, str>);
#[derive(Debug)]
pub struct TransformedFlowFile<'a> {
- target_relationship_name: &'static str,
+ target_relationship_name: Cow<'static, str>,
new_content: Option<Content<'a>>, // If None, the content doesn't change
- attributes_to_add: HashMap<String, String>,
+ attributes_to_add: Vec<FlowFileAttribute>,
}
impl<'a> TransformedFlowFile<'a> {
pub fn route_without_changes(target_relationship: &Relationship) -> Self {
+ Self::route_without_changes_by_name(Cow::Borrowed(target_relationship.name))
+ }
+
+ pub fn route_without_changes_by_name(relationship: Cow<'static, str>) -> Self {
Self {
- target_relationship_name: target_relationship.name,
+ target_relationship_name: relationship,
new_content: None,
- attributes_to_add: HashMap::new(),
+ attributes_to_add: Vec::new(),
}
}
- pub fn new(
- target_relationship: &Relationship,
- new_content: Option<Vec<u8>>,
- attributes_to_add: HashMap<String, String>,
- ) -> Self {
+ pub fn new(target_relationship: &Relationship, new_content: Option<Vec<u8>>) -> Self {
Self {
- target_relationship_name: target_relationship.name,
+ target_relationship_name: Cow::Borrowed(target_relationship.name),
new_content: new_content.map(Content::Buffer),
- attributes_to_add,
+ attributes_to_add: Vec::new(),
}
}
+ #[must_use]
+ pub fn with_content(mut self, content: Vec<u8>) -> Self {
+ self.new_content = Some(Content::Buffer(content));
+ self
+ }
+
pub fn new_content(&'_ self) -> Option<&'_ Content<'_>> {
self.new_content.as_ref()
}
- pub fn target_relationship(&self) -> &'static str {
- self.target_relationship_name
+ pub fn target_relationship(&self) -> &str {
+ &self.target_relationship_name
}
- pub fn attributes_to_add(&self) -> &HashMap<String, String> {
+ pub fn attributes_to_add(&self) -> &[FlowFileAttribute] {
&self.attributes_to_add
}
+ pub fn attribute(&self, name: &str) -> Option<&str> {
+ self.attributes_to_add
+ .iter()
+ .find(|(k, _)| k == name)
+ .map(|(_, v)| v.as_ref())
+ }
+
#[cfg(any(test, feature = "test-utils"))]
pub fn into_bytes(self) -> std::io::Result<Option<Vec<u8>>> {
match self.new_content {
@@ -82,6 +99,8 @@ impl<'a> TransformedFlowFile<'a> {
}
}
+impl_with_attributes!(TransformedFlowFile<'a>);
+
pub trait FlowFileTransform {
fn transform<
'a,
@@ -92,7 +111,7 @@ pub trait FlowFileTransform {
context: &Context,
input_stream: &'a mut dyn InputStream,
logger: &LoggerImpl,
- ) -> Result<TransformedFlowFile<'a>, MinifiError>;
+ ) -> Result<TransformedFlowFile<'a>, ProcessError>;
}
pub trait MutFlowFileTransform {
@@ -105,7 +124,7 @@ pub trait MutFlowFileTransform {
context: &Context,
input_stream: &'a mut dyn InputStream,
logger: &LoggerImpl,
- ) -> Result<TransformedFlowFile<'a>, MinifiError>;
+ ) -> Result<TransformedFlowFile<'a>, ProcessError>;
}
pub struct FlowFileTransformProcessorType {}
@@ -115,7 +134,7 @@ fn handle_transform<PC, PS, L, F>(
session: &mut PS,
logger: &L,
mut transform_fn: F,
-) -> Result<OnTriggerResult, MinifiError>
+) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -123,13 +142,22 @@ where
F: for<'stream> FnMut(
&ContextSessionFlowFileBundle<'_, PC, PS>,
&'stream mut dyn InputStream,
- ) -> Result<TransformedFlowFile<'stream>, MinifiError>,
+ ) -> Result<TransformedFlowFile<'stream>, ProcessError>,
{
if let Some(mut flow_file) = session.get() {
let simple_context = ContextSessionFlowFileBundle::new(context, session, Some(&flow_file));
let (attrs_to_add, relationship) = session.read_stream(&flow_file, |input_stream| {
- let transformed = transform_fn(&simple_context, input_stream)?;
+ let transformed = match transform_fn(&simple_context, input_stream) {
+ Ok(transform_success) => transform_success,
+ Err(ProcessError::Route(route)) => {
+ route.log(logger);
+ TransformedFlowFile::route_without_changes_by_name(route.relationship)
+ }
+ Err(ProcessError::Fatal(e)) => {
+ return Err(e);
+ }
+ };
info!(logger, "{:?}", transformed);
match transformed.new_content {
@@ -152,7 +180,7 @@ where
session.set_attribute(&mut flow_file, &k, &v)?;
}
- session.transfer(flow_file, relationship)?;
+ session.transfer(flow_file, relationship.as_ref())?;
Ok(OnTriggerResult::Ok)
} else {
logger.log(LogLevel::Trace, format_args!("No flowfile to transform"));
@@ -170,7 +198,7 @@ where
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -180,9 +208,7 @@ where
scheduled_impl.transform(ctx, input, &self.logger)
})
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
@@ -197,7 +223,7 @@ where
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -207,48 +233,118 @@ where
scheduled_impl.transform(ctx, input, &self.logger)
})
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
-#[macro_export]
-macro_rules! unwrap_or_route {
- ($result:expr, $route:expr) => {
- match $result {
- Ok(v) => v,
- Err(_e) => {
- return Ok(TransformedFlowFile::route_without_changes($route));
- }
- }
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::api::RawProcessor;
+ use crate::api::raw_processor::MultiThreadedTrigger;
+ use crate::{
+ GetControllerService, GetId, MockFlowFile, MockLogger, MockProcessContext,
+ MockProcessSession, ProcessError, RouteErrorExt,
};
- ($result:expr, $route:expr, $custom_logger:expr) => {
- match $result {
- Ok(v) => v,
- Err(e) => {
- minifi_native::error!(
- $custom_logger,
- "Failed to unwrap due to {}. Routing flow file.",
- e
- );
- return Ok(TransformedFlowFile::route_without_changes($route));
- }
+ struct RouteToFailure;
+ impl Schedule for RouteToFailure {
+ fn schedule<Ctx: GetProperty, L: Logger>(_c: &Ctx, _l: &L) -> Result<Self, MinifiError> {
+ Ok(RouteToFailure)
}
- };
+ }
+ impl FlowFileTransform for RouteToFailure {
+ fn transform<
+ 'a,
+ Context: GetProperty + GetControllerService + GetAttribute + GetId,
+ LoggerImpl: Logger,
+ >(
+ &self,
+ _context: &Context,
+ _input_stream: &'a mut dyn InputStream,
+ _logger: &LoggerImpl,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ let bad: Result<TransformedFlowFile<'a>, std::io::Error> = Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "bad data",
+ ));
+ bad.route_err_to_failure()
+ }
+ }
- ($result:expr, $route:expr, $custom_logger:expr, $context:expr) => {
- match $result {
- Ok(v) => v,
- Err(e) => {
- error!(
- $custom_logger,
- "Failed to {} due to {}. Routing flow file.", $context, e
- );
- return Ok(TransformedFlowFile::route_without_changes($route));
- }
+ struct FatalTransform;
+ impl Schedule for FatalTransform {
+ fn schedule<Ctx: GetProperty, L: Logger>(_c: &Ctx, _l: &L) -> Result<Self, MinifiError> {
+ Ok(FatalTransform)
}
- };
+ }
+ impl FlowFileTransform for FatalTransform {
+ fn transform<
+ 'a,
+ Context: GetProperty + GetControllerService + GetAttribute + GetId,
+ LoggerImpl: Logger,
+ >(
+ &self,
+ _context: &Context,
+ _input_stream: &'a mut dyn InputStream,
+ _logger: &LoggerImpl,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ Err(ProcessError::Fatal(MinifiError::custom("real error")))
+ }
+ }
+
+ fn seeded_session() -> MockProcessSession {
+ let mut session = MockProcessSession::new();
+ session
+ .input_flow_files
+ .push(MockFlowFile::with_content(b"data"));
+ session
+ }
+
+ #[test]
+ fn route_error_transfers_to_failure_and_commits() {
+ let mut processor: Processor<
+ RouteToFailure,
+ FlowFileTransformProcessorType,
+ MultiThreaded,
+ MockLogger,
+ > = Processor::new(MockLogger::new());
+ processor.scheduled_impl = Some(RouteToFailure);
+
+ let mut context = MockProcessContext::new();
+ let mut session = seeded_session();
+
+ let result = MultiThreadedTrigger::trigger(&processor, &mut context, &mut session);
+
+ assert_eq!(
+ result.expect("should commit, not roll back"),
+ OnTriggerResult::Ok
+ );
+ let transferred = session.transferred_flow_files.borrow();
+ assert_eq!(transferred.len(), 1);
+ assert_eq!(transferred[0].relationship, "failure");
+ }
+
+ #[test]
+ fn fatal_error_propagates_and_transfers_nothing() {
+ let mut processor: Processor<
+ FatalTransform,
+ FlowFileTransformProcessorType,
+ MultiThreaded,
+ MockLogger,
+ > = Processor::new(MockLogger::new());
+ processor.scheduled_impl = Some(FatalTransform);
+
+ let mut context = MockProcessContext::new();
+ let mut session = seeded_session();
+
+ let result = MultiThreadedTrigger::trigger(&processor, &mut context, &mut session);
+
+ assert!(matches!(
+ result,
+ Err(ProcessError::Fatal(MinifiError::CustomError(_)))
+ ));
+ assert_eq!(session.num_of_transferred_flow_files(), 0);
+ }
}
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs
index e22b3bab9..26b98128c 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs
@@ -17,3 +17,4 @@
pub(crate) mod context_session_flowfile_bundle;
pub(crate) mod flow_file_content;
+pub(crate) mod with_attributes;
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs
new file mode 100644
index 000000000..20e541c34
--- /dev/null
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs
@@ -0,0 +1,30 @@
+#[macro_export]
+macro_rules! impl_with_attributes {
+ ($name:ident $(<$lt:lifetime>)?) => {
+ impl $(<$lt>)? $name $(<$lt>)? {
+ #[must_use]
+ pub fn with_attribute(
+ mut self,
+ key: impl Into<std::borrow::Cow<'static, str>>,
+ value: impl Into<std::borrow::Cow<'static, str>>,
+ ) -> Self {
+ self.attributes_to_add.push((key.into(), value.into()).into());
+ self
+ }
+
+ #[must_use]
+ pub fn with_attributes<K, V>(
+ mut self,
+ attributes: impl IntoIterator<Item = (K, V)>
+ ) -> Self
+ where
+ K: Into<std::borrow::Cow<'static, str>>,
+ V: Into<std::borrow::Cow<'static, str>>,
+ {
+ self.attributes_to_add
+ .extend(attributes.into_iter().map(|(k, v)| (k.into(), v.into()).into()));
+ self
+ }
+ }
+ };
+}
diff --git a/minifi_rust/minifi_native/src/api/property.rs b/minifi_rust/minifi_native/src/api/property.rs
index f0c95767f..f9fc011fc 100644
--- a/minifi_rust/minifi_native/src/api/property.rs
+++ b/minifi_rust/minifi_native/src/api/property.rs
@@ -26,7 +26,7 @@ use std::marker::PhantomData;
use std::str::FromStr;
use std::time::Duration;
-#[derive(Debug, Eq, PartialEq)]
+#[derive(Debug, Eq, PartialEq, Clone)]
pub enum StandardPropertyValidator {
NonBlankValidator,
TimePeriodValidator,
@@ -38,13 +38,14 @@ pub enum StandardPropertyValidator {
F64Validator,
}
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Clone)]
pub enum PropertyConstraints {
Validator(StandardPropertyValidator),
AllowedValues(&'static [&'static str]),
ControllerService(&'static str),
}
+#[derive(Clone)]
pub struct PropertyDefinition {
pub name: &'static str,
pub description: &'static str,
@@ -151,7 +152,7 @@ impl<T: PropertyType> PropertyValue for T {
fn from_raw(raw: Option<String>, name: &str) -> Result<Self::Output, MinifiError> {
match raw {
Some(value) => T::parse(&value),
- None => Err(MinifiError::missing_required_property(name.to_string())),
+ None => Err(MinifiError::MissingRequiredProperty(name.to_owned().into())),
}
}
}
@@ -274,7 +275,7 @@ where
type Cs = Cs;
type Output<'a> = &'a Cs;
fn from_service<'a>(service: Option<&'a Cs>, name: &str) -> Result<&'a Cs, MinifiError> {
- service.ok_or_else(|| MinifiError::missing_required_property(name.to_string()))
+ service.ok_or_else(|| MinifiError::MissingRequiredProperty(name.to_owned().into()))
}
}
diff --git a/minifi_rust/minifi_native/src/api/raw_processor.rs b/minifi_rust/minifi_native/src/api/raw_processor.rs
index 51a4a0550..9ae3c8ccc 100644
--- a/minifi_rust/minifi_native/src/api/raw_processor.rs
+++ b/minifi_rust/minifi_native/src/api/raw_processor.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::api::errors::MinifiError;
-use crate::{LogLevel, Logger, ProcessContext, ProcessSession};
+use crate::{LogLevel, Logger, ProcessContext, ProcessError, ProcessSession};
pub enum ProcessorInputRequirement {
Required,
@@ -69,7 +69,7 @@ pub trait SingleThreadedTrigger: RawProcessor<Threading = SingleThreaded> {
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>;
@@ -80,7 +80,7 @@ pub trait MultiThreadedTrigger: RawProcessor<Threading = MultiThreaded> {
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>;
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 49df13f24..a2207ca85 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
@@ -517,11 +517,14 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
let mut reader = CffiInputStream::new(stream_ptr);
if let Some(cb) = ctx.callback.take() {
- match cb(&mut reader) {
- Ok(cb_ok) => ctx.result = Some(Ok(cb_ok)),
- Err(_) => {
- return minifi_io_status_MINIFI_IO_ERROR;
- }
+ let is_err = {
+ let outcome = cb(&mut reader);
+ let is_err = outcome.is_err();
+ ctx.result = Some(outcome);
+ is_err
+ };
+ if is_err {
+ return minifi_io_status_MINIFI_IO_ERROR;
}
} else {
return minifi_io_status_MINIFI_IO_ERROR;
@@ -538,6 +541,10 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
&mut ctx as *mut _ as *mut c_void,
);
+ if let Some(result) = ctx.result.take() {
+ return result;
+ }
+
if status != minifi_status_MINIFI_STATUS_SUCCESS {
return Err(MinifiError::StatusError((
"minifi_process_session_read".into(),
@@ -545,11 +552,7 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
)));
}
- if let Some(result) = ctx.result.take() {
- result
- } else {
- Err(MinifiError::UnknownError)
- }
+ Err(MinifiError::UnknownError)
}
}
diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs
index a5c6688c0..c2b02407f 100644
--- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs
+++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs
@@ -30,9 +30,37 @@ use crate::{
ComponentIdentifier, LogLevel, MultiThreaded, OutputAttribute, Processor, ProcessorDefinition,
PropertyDefinition, Schedule, SingleThreaded,
};
-use crate::{OnTriggerResult, Relationship};
+use crate::{OnTriggerResult, ProcessError, Relationship};
use minifi_native_sys::*;
+fn process_error_to_status<P: RawProcessor>(
+ processor: &P,
+ result: Result<OnTriggerResult, ProcessError>,
+) -> minifi_status {
+ match result {
+ Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
+ Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
+ Err(ProcessError::Fatal(err)) => {
+ processor.log(
+ LogLevel::Error,
+ format_args!("Error during trigger {}", err),
+ );
+ err.to_status()
+ }
+ Err(ProcessError::Route(route)) => {
+ processor.log(
+ LogLevel::Warn,
+ format_args!(
+ "Cannot route to '{}' at the top level (no current flow file); \
+ failing the trigger: {}",
+ route.relationship, route.source
+ ),
+ );
+ minifi_status_MINIFI_STATUS_UNKNOWN_ERROR
+ }
+ }
+}
+
pub trait DispatchOnTrigger<M: ThreadingModel> {
/// # Safety
///
@@ -57,17 +85,7 @@ where
let processor = &*(processor_ptr as *const T);
let mut context = CffiProcessContext::new(context_ptr);
let mut session = CffiProcessSession::new(session_ptr);
- match processor.trigger(&mut context, &mut session) {
- Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
- Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
- Err(minifi_error) => {
- processor.log(
- LogLevel::Error,
- format_args!("Error during trigger {}", minifi_error),
- );
- minifi_error.to_status()
- }
- }
+ process_error_to_status(processor, processor.trigger(&mut context, &mut session))
}
}
}
@@ -85,17 +103,8 @@ where
let processor = &mut *(processor_ptr as *mut T);
let mut context = CffiProcessContext::new(context_ptr);
let mut session = CffiProcessSession::new(session_ptr);
- match processor.trigger(&mut context, &mut session) {
- Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
- Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
- Err(minifi_error) => {
- processor.log(
- LogLevel::Error,
- format_args!("Error during trigger {}", minifi_error),
- );
- minifi_error.to_status()
- }
- }
+ let result = processor.trigger(&mut context, &mut session);
+ process_error_to_status(&*processor, result)
}
}
}
@@ -284,7 +293,7 @@ where
Implementation::SUPPORTS_DYNAMIC_RELATIONSHIPS,
Implementation::OUTPUT_ATTRIBUTES,
Implementation::RELATIONSHIPS,
- Implementation::PROPERTIES,
+ Implementation::properties(),
))
}
}
diff --git a/minifi_rust/minifi_native/src/lib.rs b/minifi_rust/minifi_native/src/lib.rs
index ce54f9090..b855291a2 100644
--- a/minifi_rust/minifi_native/src/lib.rs
+++ b/minifi_rust/minifi_native/src/lib.rs
@@ -20,7 +20,7 @@ mod api;
pub mod c_ffi;
pub mod mock;
-pub use api::errors::MinifiError;
+pub use api::errors::{MinifiError, ProcessError, RouteError, RouteErrorExt};
pub use api::component_definition_traits::{
ComponentIdentifier, ControllerServiceDefinition, ProcessorDefinition,
@@ -35,7 +35,7 @@ pub use api::processor_wrappers::flow_file_stream_transform::{
TransformStreamResult,
};
pub use api::processor_wrappers::flow_file_transform::{
- FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile,
+ FlowFileAttribute, FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile,
};
pub use api::processor_wrappers::utils::flow_file_content::Content;