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 <[email protected]> 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.", -);
