This is an automated email from the ASF dual-hosted git repository. martinzink pushed a commit to branch minifi_rust_pgp in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit ef46ce12c66b7386deface97b2bec0c70c0bd929 Author: Martin Zink <[email protected]> AuthorDate: Wed Jul 8 10:30:49 2026 +0200 MINIFICPP-2886 Improve rust api process errors --- .../features/error-handling.feature | 4 +- .../src/processors/asciify_german.rs | 16 +- .../asciify_german/processor_definition.rs | 4 +- .../src/processors/asciify_german/tests.rs | 7 +- .../src/processors/count_actual_logging.rs | 11 +- .../src/processors/duplicate_text.rs | 11 +- .../src/processors/generate_flow_file.rs | 8 +- .../generate_flow_file/processor_definition.rs | 17 +- .../src/processors/get_file.rs | 8 +- .../processors/get_file/processor_definition.rs | 27 +-- .../src/processors/get_file/tests.rs | 5 +- .../src/processors/kamikaze_processor.rs | 14 +- .../kamikaze_processor/processor_definition.rs | 9 +- .../src/processors/kamikaze_processor/tests.rs | 7 +- .../src/processors/log_attribute.rs | 7 +- .../log_attribute/processor_definition.rs | 21 +- .../src/processors/lorem_ipsum_cs_user.rs | 9 +- .../lorem_ipsum_cs_user/processor_definition.rs | 7 +- .../src/processors/put_file.rs | 7 +- .../processors/put_file/processor_definition.rs | 35 ++-- .../src/processors/zoo_processor.rs | 14 +- minifi_rust/minifi_native/src/api/attribute.rs | 11 +- .../src/api/component_definition_traits.rs | 3 +- minifi_rust/minifi_native/src/api/errors.rs | 226 +++++++++++++++++++-- .../api/processor_wrappers/complex_processor.rs | 18 +- .../src/api/processor_wrappers/flow_file_source.rs | 46 ++--- .../flow_file_stream_transform.rs | 74 ++++--- .../api/processor_wrappers/flow_file_transform.rs | 224 ++++++++++++++------ .../src/api/processor_wrappers/utils.rs | 1 + .../processor_wrappers/utils/with_attributes.rs | 30 +++ minifi_rust/minifi_native/src/api/property.rs | 9 +- minifi_rust/minifi_native/src/api/raw_processor.rs | 6 +- .../src/c_ffi/c_ffi_process_session.rs | 23 ++- .../src/c_ffi/c_ffi_processor_definition.rs | 57 +++--- minifi_rust/minifi_native/src/lib.rs | 4 +- 35 files changed, 663 insertions(+), 317 deletions(-) diff --git a/minifi_rust/extensions/minifi_rs_playground/features/error-handling.feature b/minifi_rust/extensions/minifi_rs_playground/features/error-handling.feature index 5eb3329f8..a9dccfd13 100644 --- a/minifi_rust/extensions/minifi_rs_playground/features/error-handling.feature +++ b/minifi_rust/extensions/minifi_rs_playground/features/error-handling.feature @@ -36,7 +36,7 @@ Feature: API error handling and logging When the MiNiFi instance starts up - Then the Minifi logs contain the following message: "KamikazeProcessorRs] [error] Error during schedule: ScheduleError("it was designed to fail during schedule")" in less than 10 seconds + Then the Minifi logs contain the following message: "KamikazeProcessorRs] [error] Error during schedule: CustomError("it was designed to fail during schedule")" in less than 10 seconds And the Minifi logs contain the following message: "(KamikazeProcessorRs): Process Schedule Operation: Error while scheduling processor" in less than 10 seconds Scenario: Minifi handles errors from trigger @@ -46,7 +46,7 @@ Feature: API error handling and logging When the MiNiFi instance starts up - Then the Minifi logs contain the following message: "KamikazeProcessorRs] [error] Error during trigger TriggerError("it was designed to fail in trigger")" in less than 10 seconds + Then the Minifi logs contain the following message: "KamikazeProcessorRs] [error] Error during trigger CustomError("it was designed to fail in trigger")" in less than 10 seconds And the Minifi logs contain the following message: "Trigger and commit failed for processor KamikazeProcessorRs" in less than 10 seconds Scenario: Panic in extension's schedule crashes the agent aswell diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs index a4d8f68ec..6d67c7d05 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german.rs @@ -20,10 +20,9 @@ use crate::processors::asciify_german::relationships::FAILURE; use minifi_native::macros::ComponentIdentifier; use minifi_native::{ - FlowFileStreamTransform, GetProperty, InputStream, Logger, MinifiError, OutputStream, Schedule, - TransformStreamResult, + FlowFileStreamTransform, GetProperty, InputStream, Logger, MinifiError, OutputStream, + ProcessError, RouteErrorExt, Schedule, TransformStreamResult, }; -use std::collections::HashMap; mod relationships; @@ -46,7 +45,7 @@ impl FlowFileStreamTransform for AsciifyGerman { input_stream: &mut dyn InputStream, output_stream: &mut dyn OutputStream, _logger: &LoggerImpl, - ) -> Result<TransformStreamResult, MinifiError> { + ) -> Result<TransformStreamResult, ProcessError> { let mut byte = [0u8; 1]; while input_stream.read(&mut byte)? > 0 { @@ -57,8 +56,8 @@ impl FlowFileStreamTransform for AsciifyGerman { 0xC3 => { let mut next = [0u8; 1]; if input_stream.read(&mut next)? == 0 { - // Truncated multi-byte sequence at EOF — treat as malformed input. - return Ok(TransformStreamResult::route_without_changes(&FAILURE)); + Err(MinifiError::custom("Truncated multi-byte sequence at EOF")) + .route_err_to_failure()? } match next[0] { 0xA4 => output_stream.write_all(b"ae")?, // ä @@ -76,10 +75,7 @@ impl FlowFileStreamTransform for AsciifyGerman { } output_stream.flush()?; - Ok(TransformStreamResult::new( - &relationships::SUCCESS, - HashMap::new(), - )) + Ok(TransformStreamResult::new(&relationships::SUCCESS)) } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs index 2a64c8a3e..4baa9d9cc 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/processor_definition.rs @@ -29,5 +29,7 @@ impl ProcessorDefinition for AsciifyGerman { const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[super::relationships::SUCCESS, super::relationships::FAILURE]; - const PROPERTIES: &'static [PropertyDefinition] = &[]; + fn properties() -> &'static [PropertyDefinition] { + &[] + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs index c9b93c658..1a2a77d60 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs @@ -83,9 +83,6 @@ fn truncated_umlaut_at_eof_routes_to_failure() { let mut input_stream = BufReader::new(input_bytes); let mut output_vec: Vec<u8> = Vec::new(); - let result = asciify_german - .transform(&context, &mut input_stream, &mut output_vec, &logger) - .expect("Should succeed"); - assert_eq!(result.write_status(), IoState::Cancel); - assert_eq!(result.target_relationship_name(), FAILURE.name); + let result = asciify_german.transform(&context, &mut input_stream, &mut output_vec, &logger); + assert!(result.is_err()); } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs index 518751d62..be37674ba 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/count_actual_logging.rs @@ -20,8 +20,8 @@ use minifi_native::macros::ComponentIdentifier; use minifi_native::{ GetProperty, Logger, MinifiError, MutTrigger, OnTriggerResult, OutputAttribute, ProcessContext, - ProcessSession, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition, - Relationship, Schedule, debug, info, trace, + ProcessError, ProcessSession, ProcessorDefinition, ProcessorInputRequirement, + PropertyDefinition, Relationship, Schedule, debug, info, trace, }; #[derive(Debug, ComponentIdentifier)] @@ -51,7 +51,7 @@ impl MutTrigger for CountActualLogging { _context: &mut PC, _session: &mut PS, logger: &L, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -72,5 +72,8 @@ impl ProcessorDefinition for CountActualLogging { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[]; - const PROPERTIES: &'static [PropertyDefinition] = &[]; + + fn properties() -> &'static [PropertyDefinition] { + &[] + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs index ee4ae055e..ee049cd4f 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/duplicate_text.rs @@ -18,10 +18,9 @@ use minifi_native::macros::ComponentIdentifier; use minifi_native::{ GetAttribute, GetControllerService, GetProperty, InputStream, Logger, MinifiError, - MutFlowFileStreamTransform, OutputAttribute, OutputStream, ProcessorDefinition, + MutFlowFileStreamTransform, OutputAttribute, OutputStream, ProcessError, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition, Relationship, Schedule, TransformStreamResult, }; -use std::collections::HashMap; #[derive(Debug, ComponentIdentifier)] pub(crate) struct DuplicateStreamText {} @@ -50,13 +49,13 @@ impl MutFlowFileStreamTransform for DuplicateStreamText { input_stream: &mut dyn InputStream, output_stream: &mut dyn OutputStream, _logger: &LoggerImpl, - ) -> Result<TransformStreamResult, MinifiError> { + ) -> Result<TransformStreamResult, ProcessError> { let mut byte = [0u8; 1]; while input_stream.read(&mut byte)? > 0 { let _ = output_stream.write(&byte)?; let _ = output_stream.write(&byte)?; } - Ok(TransformStreamResult::new(&SUCCESS, HashMap::new())) + Ok(TransformStreamResult::new(&SUCCESS)) } } @@ -67,5 +66,7 @@ impl ProcessorDefinition for DuplicateStreamText { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS]; - const PROPERTIES: &'static [PropertyDefinition] = &[]; + fn properties() -> &'static [PropertyDefinition] { + &[] + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs index e4c426fd4..8db7905d6 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs @@ -19,8 +19,8 @@ use minifi_native::macros::{ComponentIdentifier, PropertyType}; use minifi_native::{ - GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule, - Trigger, + GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError, + ProcessSession, Schedule, Trigger, }; use rand::RngExt; use rand::distr::Alphanumeric; @@ -147,7 +147,7 @@ impl Trigger for GenerateFlowFileRs { context: &mut PC, session: &mut PS, _logger: &L, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -163,7 +163,7 @@ impl Trigger for GenerateFlowFileRs { context .get_raw_property(&properties::CUSTOM_TEXT, None)? .ok_or_else(|| { - MinifiError::trigger_err( + MinifiError::custom( "GenerateFlowFile is in CustomText mode but the \"Custom Text\" \ property is not set at trigger time", ) diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs index d3f9a9391..a3e0e8a68 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs @@ -29,11 +29,14 @@ impl ProcessorDefinition for GenerateFlowFileRs { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS]; - const PROPERTIES: &'static [PropertyDefinition] = property_definitions![ - FILE_SIZE, - BATCH_SIZE, - DATA_FORMAT, - UNIQUE_FLOW_FILES, - CUSTOM_TEXT - ]; + fn properties() -> &'static [PropertyDefinition] { + const PROPERTIES: &[PropertyDefinition] = property_definitions![ + FILE_SIZE, + BATCH_SIZE, + DATA_FORMAT, + UNIQUE_FLOW_FILES, + CUSTOM_TEXT + ]; + PROPERTIES + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs index ab2f0c037..7ec9f4130 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs @@ -26,8 +26,8 @@ use crate::processors::get_file::properties::{ }; use minifi_native::macros::ComponentIdentifier; use minifi_native::{ - GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, - Schedule, Trigger, debug, info, trace, warn, + GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError, + ProcessSession, Schedule, Trigger, debug, info, trace, warn, }; use std::collections::VecDeque; use std::error; @@ -225,7 +225,7 @@ impl Schedule for GetFileRs { { let input_directory = context.get_property(&DIRECTORY)?; if !input_directory.is_dir() { - return Err(MinifiError::schedule_err(format!( + return Err(MinifiError::custom(format!( "{:?} is not a valid directory", input_directory ))); @@ -269,7 +269,7 @@ impl Trigger for GetFileRs { context: &mut PC, session: &mut PS, logger: &L, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs index 59171d3d0..13dbbbbdc 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs @@ -33,16 +33,19 @@ impl ProcessorDefinition for GetFileRs { const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE]; const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS]; - const PROPERTIES: &'static [PropertyDefinition] = property_definitions![ - DIRECTORY, - POLLING_INTERVAL, - RECURSE, - KEEP_SOURCE_FILE, - MIN_AGE, - MAX_AGE, - MIN_SIZE, - MAX_SIZE, - IGNORE_HIDDEN_FILES, - BATCH_SIZE, - ]; + fn properties() -> &'static [PropertyDefinition] { + const PROPERTIES: &[PropertyDefinition] = property_definitions![ + DIRECTORY, + POLLING_INTERVAL, + RECURSE, + KEEP_SOURCE_FILE, + MIN_AGE, + MAX_AGE, + MIN_SIZE, + MAX_SIZE, + IGNORE_HIDDEN_FILES, + BATCH_SIZE, + ]; + PROPERTIES + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs index 50691c773..d7b9ed96f 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs @@ -38,10 +38,7 @@ fn schedule_fails_with_invalid_input_dir() { "Input Directory".to_string(), "/invalid_directory".to_string(), ); - assert!(matches!( - GetFileRs::schedule(&context, &MockLogger::new()), - Err(MinifiError::ScheduleError(_)) - )); + assert!(GetFileRs::schedule(&context, &MockLogger::new()).is_err()); } #[test] diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs index f6f843576..0faa04f18 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs @@ -25,8 +25,8 @@ use crate::processors::kamikaze_processor::properties::{ }; use minifi_native::macros::{ComponentIdentifier, PropertyType}; use minifi_native::{ - GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule, - Trigger, + GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError, + ProcessSession, Schedule, Trigger, }; use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames}; @@ -57,7 +57,7 @@ impl Schedule for KamikazeProcessorRs { let schedule_behaviour = context.get_property(&SCHEDULE_BEHAVIOUR)?; match schedule_behaviour { - KamikazeBehaviour::ReturnErr => Err(MinifiError::schedule_err( + KamikazeBehaviour::ReturnErr => Err(MinifiError::custom( "it was designed to fail during schedule", )), KamikazeBehaviour::ReturnOk => Ok(KamikazeProcessorRs { trigger_behaviour }), @@ -81,16 +81,16 @@ impl Trigger for KamikazeProcessorRs { context: &mut PC, _session: &mut PS, _logger: &L, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, L: Logger, { match self.trigger_behaviour { - KamikazeBehaviour::ReturnErr => Err(MinifiError::trigger_err( - "it was designed to fail in trigger", - )), + KamikazeBehaviour::ReturnErr => { + Err(MinifiError::custom("it was designed to fail in trigger").into()) + } KamikazeBehaviour::ReturnOk => Ok(OnTriggerResult::Ok), KamikazeBehaviour::Panic => { panic!("KamikazeProcessor::trigger panic") diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs index 51c8ff0c8..f9a47d115 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs @@ -28,8 +28,9 @@ impl ProcessorDefinition for KamikazeProcessorRs { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS]; - const PROPERTIES: &'static [PropertyDefinition] = property_definitions![ - properties::SCHEDULE_BEHAVIOUR, - properties::TRIGGER_BEHAVIOUR - ]; + fn properties() -> &'static [PropertyDefinition] { + const PROPERTIES: &[PropertyDefinition] = + property_definitions![SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR]; + PROPERTIES + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs index dd6e4794f..2b7f52166 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs @@ -17,8 +17,7 @@ use super::*; use crate::processors::kamikaze_processor::properties::{SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR}; -use minifi_native::MinifiError::{ScheduleError, TriggerError}; -use minifi_native::{MockLogger, MockProcessContext, MockProcessSession}; +use minifi_native::{MockLogger, MockProcessContext, MockProcessSession, ProcessError}; use std::panic::AssertUnwindSafe; #[test] @@ -36,7 +35,7 @@ fn on_schedule_err() { "ReturnErr".to_string(), ); let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new()); - assert!(matches!(processor, Err(ScheduleError(_)))); + assert!(matches!(processor, Err(MinifiError::CustomError(_)))); } #[test] @@ -78,7 +77,7 @@ fn on_trigger_err() { let mut session = MockProcessSession::new(); assert!(matches!( processor.trigger(&mut context, &mut session, &MockLogger::new()), - Err(TriggerError(_)) + Err(ProcessError::Fatal(MinifiError::CustomError(_))) )); } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs index 48de81360..329e281f4 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs @@ -21,8 +21,9 @@ use crate::processors::log_attribute::properties::{FLOW_FILES_TO_LOG, LOG_LEVEL, use minifi_native::StandardPropertyValidator::NonBlankValidator; use minifi_native::macros::ComponentIdentifier; use minifi_native::{ - GetProperty, LogLevel, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, - PropertyConstraints, PropertySchema, PropertyType, Schedule, Trigger, debug, log, trace, + GetProperty, LogLevel, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError, + ProcessSession, PropertyConstraints, PropertySchema, PropertyType, Schedule, Trigger, debug, + log, trace, }; mod properties; @@ -104,7 +105,7 @@ impl Trigger for LogAttributeRs { _context: &mut PC, session: &mut PS, logger: &L, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs index b5732645d..91e0b3fed 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs @@ -30,13 +30,16 @@ impl ProcessorDefinition for LogAttributeRs { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS]; - const PROPERTIES: &'static [PropertyDefinition] = property_definitions![ - LOG_LEVEL, - ATTRIBUTES_TO_LOG, - ATTRIBUTES_TO_IGNORE, - LOG_PAYLOAD, - LOG_PREFIX, - FLOW_FILES_TO_LOG, - HEX_ENCODE_PAYLOAD, - ]; + fn properties() -> &'static [PropertyDefinition] { + const PROPERTIES: &[PropertyDefinition] = property_definitions![ + LOG_LEVEL, + ATTRIBUTES_TO_LOG, + ATTRIBUTES_TO_IGNORE, + LOG_PAYLOAD, + LOG_PREFIX, + FLOW_FILES_TO_LOG, + HEX_ENCODE_PAYLOAD, + ]; + PROPERTIES + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs index f8a684d9f..4d32739e2 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs @@ -25,9 +25,8 @@ use crate::processors::lorem_ipsum_cs_user::relationships::SUCCESS; use minifi_native::macros::{ComponentIdentifier, PropertyType}; use minifi_native::{ Content, FlowFileSource, GeneratedFlowFile, GetControllerService, GetProperty, Logger, - MinifiError, Schedule, trace, + MinifiError, ProcessError, Schedule, trace, }; -use std::collections::HashMap; use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames}; #[derive( @@ -59,7 +58,7 @@ impl FlowFileSource for LoremIpsumCSUser { &self, context: &'a mut Context, logger: &LoggerImpl, - ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError> { + ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError> { trace!(logger, "generate call {:?}", self); let dummy_controller_service = context.get_controller_service(&DUMMY_CONTROLLER_SERVICE)?; trace!( @@ -72,15 +71,13 @@ impl FlowFileSource for LoremIpsumCSUser { let generated_flow_file = GeneratedFlowFile::new( &SUCCESS, Some(Content::from(controller_service.data.clone())), - HashMap::new(), ); Ok(vec![generated_flow_file]) } WriteMethod::Stream => { let reader = controller_service.data.as_bytes(); let content = Content::Stream(Box::new(reader)); - let generated_flow_file = - GeneratedFlowFile::new(&SUCCESS, Some(content), HashMap::new()); + let generated_flow_file = GeneratedFlowFile::new(&SUCCESS, Some(content)); Ok(vec![generated_flow_file]) } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs index 78c09ef0c..a80605035 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs @@ -31,6 +31,9 @@ impl ProcessorDefinition for LoremIpsumCSUser { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS]; - const PROPERTIES: &'static [PropertyDefinition] = - property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD]; + fn properties() -> &'static [PropertyDefinition] { + const PROPERTIES: &[PropertyDefinition] = + property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD]; + PROPERTIES + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs index e81c861ca..1dd63ec2d 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs @@ -22,7 +22,7 @@ use crate::processors::put_file::unix_permissions::PutFileUnixPermissions; use minifi_native::macros::{ComponentIdentifier, PropertyType}; use minifi_native::{ FlowFileTransform, GetAttribute, GetControllerService, GetId, GetProperty, InputStream, Logger, - MinifiError, Schedule, TransformedFlowFile, trace, unwrap_or_route, warn, + MinifiError, ProcessError, RouteErrorExt, Schedule, TransformedFlowFile, trace, warn, }; use std::path::{Path, PathBuf}; use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames}; @@ -171,11 +171,10 @@ impl FlowFileTransform for PutFileRs { context: &Context, input_stream: &'a mut dyn InputStream, logger: &LoggerImpl, - ) -> Result<TransformedFlowFile<'a>, MinifiError> { + ) -> Result<TransformedFlowFile<'a>, ProcessError> { trace!(logger, "on_trigger: {:?}", self); - let destination_path = - unwrap_or_route!(Self::get_destination_path(context), &FAILURE, logger); + let destination_path = Self::get_destination_path(context).route_err_to_failure()?; if self.directory_is_full(&destination_path) { warn!(logger, "Directory is full"); diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs index 93be840a4..49243e5c9 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs @@ -30,21 +30,24 @@ impl ProcessorDefinition for PutFileRs { const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS, FAILURE]; - #[cfg(windows)] - const PROPERTIES: &'static [PropertyDefinition] = property_definitions![ - properties::DIRECTORY, - properties::CONFLICT_RESOLUTION, - properties::CREATE_DIRS, - properties::MAX_FILE_COUNT, - ]; + 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 + }); - #[cfg(unix)] - const PROPERTIES: &'static [PropertyDefinition] = property_definitions![ - properties::DIRECTORY, - properties::CONFLICT_RESOLUTION, - properties::CREATE_DIRS, - properties::MAX_FILE_COUNT, - unix_only_properties::PERMISSIONS, - unix_only_properties::DIRECTORY_PERMISSIONS, - ]; + &COMBINED_PROPERTIES + } } diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs index 86d67c779..6eabf7a4d 100644 --- a/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs +++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs @@ -21,8 +21,8 @@ use crate::controller_services::animal_controller_apis::{ use minifi_native::macros::ComponentIdentifier; use minifi_native::{ GetProperty, Logger, MinifiError, OnTriggerResult, OutputAttribute, ProcessContext, - ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property, PropertyDefinition, - Relationship, Schedule, Trigger, critical, info, property_definitions, + ProcessError, ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property, + PropertyDefinition, Relationship, Schedule, Trigger, critical, info, property_definitions, }; pub(crate) const CAN_FLY_SERVICE: Property<dyn CanFlyControllerApi> = @@ -52,7 +52,7 @@ impl Trigger for ZooProcessorRs { context: &mut Context, _session: &mut Session, logger: &Lggr, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where Context: ProcessContext, Session: ProcessSession<FlowFile = Context::FlowFile>, @@ -81,6 +81,10 @@ impl ProcessorDefinition for ZooProcessorRs { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[]; const RELATIONSHIPS: &'static [Relationship] = &[]; - const PROPERTIES: &'static [PropertyDefinition] = - property_definitions![CAN_FLY_SERVICE, NUMBER_OF_LEGS]; + fn properties() -> &'static [PropertyDefinition] { + const PROPERTIES: &[PropertyDefinition] = + property_definitions![CAN_FLY_SERVICE, NUMBER_OF_LEGS]; + + PROPERTIES + } } diff --git a/minifi_rust/minifi_native/src/api/attribute.rs b/minifi_rust/minifi_native/src/api/attribute.rs index 571ea7ce7..ed0301654 100644 --- a/minifi_rust/minifi_native/src/api/attribute.rs +++ b/minifi_rust/minifi_native/src/api/attribute.rs @@ -16,6 +16,7 @@ // under the License. use crate::MinifiError; +use std::borrow::Cow; pub struct OutputAttribute { pub name: &'static str, @@ -23,10 +24,18 @@ pub struct OutputAttribute { pub description: &'static str, } +impl From<&OutputAttribute> for Cow<'static, str> { + fn from(attr: &OutputAttribute) -> Self { + Cow::Borrowed(attr.name) + } +} + pub trait GetAttribute { fn get_attribute(&self, name: &str) -> Result<Option<String>, MinifiError>; fn get_required_attribute(&self, name: &str) -> Result<String, MinifiError> { self.get_attribute(name)? - .ok_or(MinifiError::missing_required_attribute(name.to_owned())) + .ok_or(MinifiError::MissingRequiredAttribute( + name.to_owned().into(), + )) } } diff --git a/minifi_rust/minifi_native/src/api/component_definition_traits.rs b/minifi_rust/minifi_native/src/api/component_definition_traits.rs index 5e0e8a126..c1f473f42 100644 --- a/minifi_rust/minifi_native/src/api/component_definition_traits.rs +++ b/minifi_rust/minifi_native/src/api/component_definition_traits.rs @@ -31,7 +31,8 @@ pub trait ProcessorDefinition { const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool; const OUTPUT_ATTRIBUTES: &'static [OutputAttribute]; const RELATIONSHIPS: &'static [Relationship]; - const PROPERTIES: &'static [PropertyDefinition]; + /// fn to allow concatenation for complex processors + fn properties() -> &'static [PropertyDefinition]; } pub trait ControllerServiceDefinition: Sized + 'static { diff --git a/minifi_rust/minifi_native/src/api/errors.rs b/minifi_rust/minifi_native/src/api/errors.rs index 2c833a997..7802e397a 100644 --- a/minifi_rust/minifi_native/src/api/errors.rs +++ b/minifi_rust/minifi_native/src/api/errors.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use minifi_native::{LogLevel, Relationship}; use minifi_native_sys::minifi_status; use std::borrow::Cow; use std::error::Error; @@ -35,19 +36,143 @@ pub enum ParseError { Other, } +#[derive(Debug)] +pub struct RouteError { + pub relationship: Cow<'static, str>, + pub source: Box<dyn Error + Send + Sync + 'static>, + pub log_level: LogLevel, +} + +impl RouteError { + pub(crate) fn log<L: crate::Logger>(&self, logger: &L) { + logger.log( + self.log_level, + format_args!( + "Routing flow file to '{}': {}", + self.relationship, self.source + ), + ); + } +} + +impl fmt::Display for RouteError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "route to '{}' due to: {}", + self.relationship, self.source + ) + } +} + +impl Error for RouteError {} + +#[derive(Debug)] +pub enum ProcessError { + Route(RouteError), + Fatal(MinifiError), +} + +impl From<RouteError> for ProcessError { + fn from(err: RouteError) -> Self { + ProcessError::Route(err) + } +} + +impl From<MinifiError> for ProcessError { + fn from(err: MinifiError) -> Self { + ProcessError::Fatal(err) + } +} + +macro_rules! process_error_from_fatal { + ($($t:ty),* $(,)?) => { + $( + impl From<$t> for ProcessError { + fn from(err: $t) -> Self { + ProcessError::Fatal(MinifiError::from(err)) + } + } + )* + }; +} + +process_error_from_fatal!( + std::io::Error, + strum::ParseError, + ParseBoolError, + ParseIntError, + humantime::DurationError, + byte_unit::ParseError, + NulError, + ParseFloatError, + std::convert::Infallible, +); + +impl fmt::Display for ProcessError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ProcessError::Route(err) => write!(f, "{}", err), + ProcessError::Fatal(err) => write!(f, "{}", err), + } + } +} + +impl Error for ProcessError {} + +pub trait RouteErrorExt<T> { + fn route_err(self, rel: &Relationship, level: LogLevel) -> Result<T, ProcessError>; + + fn route_to<S: Into<Cow<'static, str>>>( + self, + relationship: S, + level: LogLevel, + ) -> Result<T, ProcessError>; + + fn route_err_to_failure(self) -> Result<T, ProcessError>; +} + +impl<T, E> RouteErrorExt<T> for Result<T, E> +where + E: Into<Box<dyn Error + Send + Sync + 'static>>, +{ + fn route_err(self, rel: &Relationship, level: LogLevel) -> Result<T, ProcessError> { + self.route_to(rel.name, level) + } + + fn route_to<S: Into<Cow<'static, str>>>( + self, + relationship: S, + level: LogLevel, + ) -> Result<T, ProcessError> { + self.map_err(|e| { + ProcessError::Route(RouteError { + relationship: relationship.into(), + source: e.into(), + log_level: level, + }) + }) + } + + fn route_err_to_failure(self) -> Result<T, ProcessError> { + self.route_to("failure", LogLevel::Warn) + } +} + #[derive(Debug)] pub enum MinifiError { UnknownError, StatusError((Cow<'static, str>, NonZeroU32)), MissingRequiredAttribute(Cow<'static, str>), MissingRequiredProperty(Cow<'static, str>), - ControllerServiceError(Cow<'static, str>), + UnscheduledProcessor, ValidationError(Cow<'static, str>), - ScheduleError(Cow<'static, str>), - TriggerError(Cow<'static, str>), + CustomError(Cow<'static, str>), Parse(ParseError), MissingFlowFileError, IoError(std::io::Error), + + Other(Box<dyn Error + Send + Sync + 'static>), } impl From<std::io::Error> for MinifiError { @@ -124,33 +249,24 @@ impl MinifiError { } } - pub fn validation_err<S: Into<Cow<'static, str>>>(msg: S) -> Self { + pub fn validation<S: Into<Cow<'static, str>>>(msg: S) -> Self { MinifiError::ValidationError(msg.into()) } - pub fn schedule_err<S: Into<Cow<'static, str>>>(msg: S) -> Self { - MinifiError::ScheduleError(msg.into()) - } - - pub fn trigger_err<S: Into<Cow<'static, str>>>(msg: S) -> Self { - MinifiError::TriggerError(msg.into()) - } - - pub fn missing_required_property<S: Into<Cow<'static, str>>>(msg: S) -> Self { - MinifiError::MissingRequiredProperty(msg.into()) - } - - pub fn missing_required_attribute<S: Into<Cow<'static, str>>>(msg: S) -> Self { - MinifiError::MissingRequiredAttribute(msg.into()) - } - - pub fn controller_service_err<S: Into<Cow<'static, str>>>(msg: S) -> Self { - MinifiError::ControllerServiceError(msg.into()) + pub fn custom<S: Into<Cow<'static, str>>>(msg: S) -> Self { + MinifiError::CustomError(msg.into()) } pub fn parse_err() -> Self { MinifiError::Parse(ParseError::Other) } + + pub fn other<E>(err: E) -> Self + where + E: Into<Box<dyn Error + Send + Sync + 'static>>, + { + MinifiError::Other(err.into()) + } } impl fmt::Display for MinifiError { @@ -177,9 +293,75 @@ impl fmt::Display for MinifiError { } _ => write!(f, "{} (Unknown Status Code: {})", context, code), }, + MinifiError::Other(err) => write!(f, "Custom error: {}", err), _ => write!(f, "{:?}", self), } } } impl Error for MinifiError {} + +#[cfg(test)] +mod tests { + use super::*; + + fn io_err() -> std::io::Error { + std::io::Error::other("boom") + } + + #[test] + fn route_err_to_failure_uses_warn() { + let res: Result<(), std::io::Error> = Err(io_err()); + match res.route_err_to_failure() { + Err(ProcessError::Route(route)) => { + assert_eq!(route.relationship.as_ref(), "failure"); + assert_eq!(route.log_level, LogLevel::Warn); + assert_eq!(route.source.to_string(), "boom"); + } + other => panic!("expected a route error, got {other:?}"), + } + } + + #[test] + fn route_err_uses_the_relationships_name() { + const REJECT: Relationship = Relationship { + name: "reject", + description: "", + }; + let res: Result<(), std::io::Error> = Err(io_err()); + match res.route_err(&REJECT, LogLevel::Info) { + Err(ProcessError::Route(route)) => { + assert_eq!(route.relationship.as_ref(), "reject"); + assert_eq!(route.log_level, LogLevel::Info); + } + other => panic!("expected a route error, got {other:?}"), + } + } + + #[test] + fn ok_values_pass_through_unchanged() { + let res: Result<u8, std::io::Error> = Ok(5); + assert_eq!(res.route_err_to_failure().unwrap(), 5); + } + + #[test] + fn minifi_error_converts_to_fatal_via_from() { + let pe: ProcessError = MinifiError::custom("nope").into(); + assert!(matches!( + pe, + ProcessError::Fatal(MinifiError::CustomError(_)) + )); + } + + #[test] + fn raw_error_question_mark_becomes_fatal() { + fn inner() -> Result<(), ProcessError> { + Err(io_err())?; + Ok(()) + } + assert!(matches!( + inner(), + Err(ProcessError::Fatal(MinifiError::IoError(_))) + )); + } +} diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs index 24f29384f..7699f4b33 100644 --- a/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs +++ b/minifi_rust/minifi_native/src/api/processor_wrappers/complex_processor.rs @@ -18,7 +18,7 @@ use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger}; use crate::{ ComponentIdentifier, Logger, MinifiError, MultiThreaded, OnTriggerResult, ProcessContext, - ProcessSession, Processor, ProcessorDefinition, Schedule, SingleThreaded, + ProcessError, ProcessSession, Processor, ProcessorDefinition, Schedule, SingleThreaded, }; pub trait MutTrigger { @@ -27,7 +27,7 @@ pub trait MutTrigger { context: &mut Ctx, session: &mut Session, logger: &Lggr, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where Ctx: ProcessContext, Session: ProcessSession<FlowFile = Ctx::FlowFile>, @@ -40,7 +40,7 @@ pub trait Trigger { context: &mut Context, session: &mut Session, logger: &Lggr, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where Context: ProcessContext, Session: ProcessSession<FlowFile = Context::FlowFile>, @@ -59,7 +59,7 @@ where &mut self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -67,9 +67,7 @@ where if let Some(ref mut scheduled_impl) = self.scheduled_impl { scheduled_impl.trigger(context, session, &self.logger) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } @@ -84,7 +82,7 @@ where &self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -92,9 +90,7 @@ where if let Some(ref scheduled_impl) = self.scheduled_impl { scheduled_impl.trigger(context, session, &self.logger) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs index 1f049ff12..08fd8fbf4 100644 --- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs +++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_source.rs @@ -17,42 +17,42 @@ use crate::api::processor_wrappers::utils::flow_file_content::Content; use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger}; +use crate::{FlowFileAttribute, impl_with_attributes}; use crate::{ GetControllerService, GetProperty, Logger, MinifiError, MultiThreaded, OnTriggerResult, - ProcessContext, ProcessSession, Processor, Relationship, Schedule, SingleThreaded, + ProcessContext, ProcessError, ProcessSession, Processor, Relationship, Schedule, + SingleThreaded, }; -use std::collections::HashMap; +use std::borrow::Cow; pub struct GeneratedFlowFile<'a> { - target_relationship_name: &'static str, + target_relationship_name: Cow<'static, str>, new_content: Option<Content<'a>>, - attributes_to_add: HashMap<String, String>, + attributes_to_add: Vec<FlowFileAttribute>, } impl<'a> GeneratedFlowFile<'a> { - pub fn new( - target_relationship: &'a Relationship, - new_content: Option<Content<'a>>, - attributes_to_add: HashMap<String, String>, - ) -> Self { + pub fn new(target_relationship: &'a Relationship, new_content: Option<Content<'a>>) -> Self { Self { - target_relationship_name: target_relationship.name, + target_relationship_name: Cow::Borrowed(target_relationship.name), new_content, - attributes_to_add, + attributes_to_add: Vec::new(), } } - pub fn target_relationship_name(&self) -> &'static str { - self.target_relationship_name + pub fn target_relationship_name(&self) -> &str { + &self.target_relationship_name } } +impl_with_attributes!(GeneratedFlowFile<'a>); + pub trait FlowFileSource { fn generate<'a, Context: GetProperty + GetControllerService, LoggerImpl: Logger>( &self, context: &'a mut Context, logger: &LoggerImpl, - ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError>; + ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError>; } pub trait MutFlowFileSource { @@ -60,13 +60,13 @@ pub trait MutFlowFileSource { &mut self, context: &'a mut Context, logger: &LoggerImpl, - ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError>; + ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError>; } fn handle_generated_flow_files<PC, PS>( session: &mut PS, generated_flow_files: Vec<GeneratedFlowFile>, -) -> Result<OnTriggerResult, MinifiError> +) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -85,7 +85,7 @@ where for (k, v) in &new_flow_file_data.attributes_to_add { session.set_attribute(&mut ff, k, v)?; } - session.transfer(ff, new_flow_file_data.target_relationship_name)?; + session.transfer(ff, new_flow_file_data.target_relationship_name.as_ref())?; } Ok(OnTriggerResult::Ok) } @@ -102,7 +102,7 @@ where &self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -111,9 +111,7 @@ where let files = scheduled_impl.generate(context, &self.logger)?; handle_generated_flow_files::<PC, PS>(session, files) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } @@ -128,7 +126,7 @@ where &mut self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -137,9 +135,7 @@ where let files = scheduled_impl.generate(context, &self.logger)?; handle_generated_flow_files::<PC, PS>(session, files) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs index ece4244b9..5b6a2643a 100644 --- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs +++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs @@ -18,45 +18,50 @@ use crate::api::process_session::IoState; use crate::api::processor_wrappers::utils::context_session_flowfile_bundle::ContextSessionFlowFileBundle; use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger}; +use crate::{FlowFileAttribute, impl_with_attributes}; use crate::{ GetAttribute, GetControllerService, GetProperty, InputStream, LogLevel, Logger, MinifiError, - MultiThreaded, OnTriggerResult, OutputStream, ProcessContext, ProcessSession, Processor, - Relationship, Schedule, SingleThreaded, + MultiThreaded, OnTriggerResult, OutputStream, ProcessContext, ProcessError, ProcessSession, + Processor, Relationship, Schedule, SingleThreaded, }; -use std::collections::HashMap; +use std::borrow::Cow; pub struct TransformStreamResult { - target_relationship_name: &'static str, - attributes_to_add: HashMap<String, String>, + target_relationship_name: Cow<'static, str>, + attributes_to_add: Vec<FlowFileAttribute>, write_status: IoState, } impl TransformStreamResult { - pub fn new( - target_relationship: &Relationship, - attributes_to_add: HashMap<String, String>, - ) -> Self { + pub fn new(target_relationship: &Relationship) -> Self { Self { - target_relationship_name: target_relationship.name, - attributes_to_add, + target_relationship_name: Cow::Borrowed(target_relationship.name), + attributes_to_add: Vec::new(), write_status: IoState::Ok, } } pub fn route_without_changes(target_relationship: &Relationship) -> Self { + Self::route_without_changes_by_name(Cow::Borrowed(target_relationship.name)) + } + + pub fn route_without_changes_by_name(relationship: Cow<'static, str>) -> Self { Self { - target_relationship_name: target_relationship.name, - attributes_to_add: HashMap::new(), + target_relationship_name: relationship, + attributes_to_add: Vec::new(), write_status: IoState::Cancel, } } - pub fn target_relationship_name(&self) -> &'static str { - self.target_relationship_name + pub fn target_relationship_name(&self) -> &str { + &self.target_relationship_name } - pub fn get_attribute(&self, name: &str) -> Option<String> { - self.attributes_to_add.get(name).cloned() + pub fn get_attribute(&self, name: &str) -> Option<&str> { + self.attributes_to_add + .iter() + .find(|(k, _)| k == name) + .map(|(_, v)| v.as_ref()) } pub fn write_status(&self) -> IoState { @@ -64,6 +69,8 @@ impl TransformStreamResult { } } +impl_with_attributes!(TransformStreamResult); + pub trait FlowFileStreamTransform { fn transform<Ctx: GetProperty + GetControllerService + GetAttribute, LoggerImpl: Logger>( &self, @@ -71,7 +78,7 @@ pub trait FlowFileStreamTransform { input_stream: &mut dyn InputStream, output_stream: &mut dyn OutputStream, logger: &LoggerImpl, - ) -> Result<TransformStreamResult, MinifiError>; + ) -> Result<TransformStreamResult, ProcessError>; } pub trait MutFlowFileStreamTransform { @@ -81,7 +88,7 @@ pub trait MutFlowFileStreamTransform { input_stream: &mut dyn InputStream, output_stream: &mut dyn OutputStream, logger: &LoggerImpl, - ) -> Result<TransformStreamResult, MinifiError>; + ) -> Result<TransformStreamResult, ProcessError>; } pub struct FlowFileStreamTransformProcessorType {} @@ -91,7 +98,7 @@ fn handle_stream_transform<PC, PS, L, F>( session: &mut PS, logger: &L, mut transform_fn: F, -) -> Result<OnTriggerResult, MinifiError> +) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -100,14 +107,23 @@ where &ContextSessionFlowFileBundle<PC, PS>, &mut dyn InputStream, &mut dyn OutputStream, - ) -> Result<TransformStreamResult, MinifiError>, + ) -> Result<TransformStreamResult, ProcessError>, { if let Some(mut flow_file) = session.get() { let simple_context = ContextSessionFlowFileBundle::new(context, session, Some(&flow_file)); let (relationship, attrs) = session.read_stream(&flow_file, |input_stream| { session.write_stream(&flow_file, |output_stream| { - let transformed = transform_fn(&simple_context, input_stream, output_stream)?; + let transformed = match transform_fn(&simple_context, input_stream, output_stream) { + Ok(t) => t, + Err(ProcessError::Route(route)) => { + route.log(logger); + TransformStreamResult::route_without_changes_by_name(route.relationship) + } + Err(ProcessError::Fatal(e)) => { + return Err(e); + } + }; Ok(( ( @@ -123,7 +139,7 @@ where session.set_attribute(&mut flow_file, &k, &v)?; } - session.transfer(flow_file, relationship)?; + session.transfer(flow_file, relationship.as_ref())?; Ok(OnTriggerResult::Ok) } else { @@ -142,7 +158,7 @@ where &self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -152,9 +168,7 @@ where scheduled_impl.transform(ctx, input, output, &self.logger) }) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } @@ -169,7 +183,7 @@ where &mut self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -179,9 +193,7 @@ where scheduled_impl.transform(ctx, input, output, &self.logger) }) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs index 9b21b0770..462b70973 100644 --- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs +++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs @@ -15,7 +15,6 @@ // specific language governing permissions and limitations // under the License. -use crate::api::InputStream; use crate::api::flow_file::GetId; use crate::api::processor::Processor; use crate::api::processor_wrappers::utils::context_session_flowfile_bundle::ContextSessionFlowFileBundle; @@ -24,50 +23,68 @@ use crate::api::property::{GetControllerService, GetProperty}; use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger}; use crate::{ GetAttribute, LogLevel, Logger, MinifiError, MultiThreaded, OnTriggerResult, ProcessContext, - ProcessSession, Relationship, Schedule, SingleThreaded, info, + ProcessError, ProcessSession, Relationship, Schedule, SingleThreaded, impl_with_attributes, + info, }; -use std::collections::HashMap; + +use minifi_native::InputStream; +use std::borrow::Cow; + +pub type FlowFileAttribute = (Cow<'static, str>, Cow<'static, str>); #[derive(Debug)] pub struct TransformedFlowFile<'a> { - target_relationship_name: &'static str, + target_relationship_name: Cow<'static, str>, new_content: Option<Content<'a>>, // If None, the content doesn't change - attributes_to_add: HashMap<String, String>, + attributes_to_add: Vec<FlowFileAttribute>, } impl<'a> TransformedFlowFile<'a> { pub fn route_without_changes(target_relationship: &Relationship) -> Self { + Self::route_without_changes_by_name(Cow::Borrowed(target_relationship.name)) + } + + pub fn route_without_changes_by_name(relationship: Cow<'static, str>) -> Self { Self { - target_relationship_name: target_relationship.name, + target_relationship_name: relationship, new_content: None, - attributes_to_add: HashMap::new(), + attributes_to_add: Vec::new(), } } - pub fn new( - target_relationship: &Relationship, - new_content: Option<Vec<u8>>, - attributes_to_add: HashMap<String, String>, - ) -> Self { + pub fn new(target_relationship: &Relationship, new_content: Option<Vec<u8>>) -> Self { Self { - target_relationship_name: target_relationship.name, + target_relationship_name: Cow::Borrowed(target_relationship.name), new_content: new_content.map(Content::Buffer), - attributes_to_add, + attributes_to_add: Vec::new(), } } + #[must_use] + pub fn with_content(mut self, content: Vec<u8>) -> Self { + self.new_content = Some(Content::Buffer(content)); + self + } + pub fn new_content(&'_ self) -> Option<&'_ Content<'_>> { self.new_content.as_ref() } - pub fn target_relationship(&self) -> &'static str { - self.target_relationship_name + pub fn target_relationship(&self) -> &str { + &self.target_relationship_name } - pub fn attributes_to_add(&self) -> &HashMap<String, String> { + pub fn attributes_to_add(&self) -> &[FlowFileAttribute] { &self.attributes_to_add } + pub fn attribute(&self, name: &str) -> Option<&str> { + self.attributes_to_add + .iter() + .find(|(k, _)| k == name) + .map(|(_, v)| v.as_ref()) + } + #[cfg(any(test, feature = "test-utils"))] pub fn into_bytes(self) -> std::io::Result<Option<Vec<u8>>> { match self.new_content { @@ -82,6 +99,8 @@ impl<'a> TransformedFlowFile<'a> { } } +impl_with_attributes!(TransformedFlowFile<'a>); + pub trait FlowFileTransform { fn transform< 'a, @@ -92,7 +111,7 @@ pub trait FlowFileTransform { context: &Context, input_stream: &'a mut dyn InputStream, logger: &LoggerImpl, - ) -> Result<TransformedFlowFile<'a>, MinifiError>; + ) -> Result<TransformedFlowFile<'a>, ProcessError>; } pub trait MutFlowFileTransform { @@ -105,7 +124,7 @@ pub trait MutFlowFileTransform { context: &Context, input_stream: &'a mut dyn InputStream, logger: &LoggerImpl, - ) -> Result<TransformedFlowFile<'a>, MinifiError>; + ) -> Result<TransformedFlowFile<'a>, ProcessError>; } pub struct FlowFileTransformProcessorType {} @@ -115,7 +134,7 @@ fn handle_transform<PC, PS, L, F>( session: &mut PS, logger: &L, mut transform_fn: F, -) -> Result<OnTriggerResult, MinifiError> +) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -123,13 +142,22 @@ where F: for<'stream> FnMut( &ContextSessionFlowFileBundle<'_, PC, PS>, &'stream mut dyn InputStream, - ) -> Result<TransformedFlowFile<'stream>, MinifiError>, + ) -> Result<TransformedFlowFile<'stream>, ProcessError>, { if let Some(mut flow_file) = session.get() { let simple_context = ContextSessionFlowFileBundle::new(context, session, Some(&flow_file)); let (attrs_to_add, relationship) = session.read_stream(&flow_file, |input_stream| { - let transformed = transform_fn(&simple_context, input_stream)?; + let transformed = match transform_fn(&simple_context, input_stream) { + Ok(transform_success) => transform_success, + Err(ProcessError::Route(route)) => { + route.log(logger); + TransformedFlowFile::route_without_changes_by_name(route.relationship) + } + Err(ProcessError::Fatal(e)) => { + return Err(e); + } + }; info!(logger, "{:?}", transformed); match transformed.new_content { @@ -152,7 +180,7 @@ where session.set_attribute(&mut flow_file, &k, &v)?; } - session.transfer(flow_file, relationship)?; + session.transfer(flow_file, relationship.as_ref())?; Ok(OnTriggerResult::Ok) } else { logger.log(LogLevel::Trace, format_args!("No flowfile to transform")); @@ -170,7 +198,7 @@ where &self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -180,9 +208,7 @@ where scheduled_impl.transform(ctx, input, &self.logger) }) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } @@ -197,7 +223,7 @@ where &mut self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>, @@ -207,50 +233,118 @@ where scheduled_impl.transform(ctx, input, &self.logger) }) } else { - Err(MinifiError::trigger_err( - "The processor hasn't been scheduled yet", - )) + Err(MinifiError::UnscheduledProcessor.into()) } } } -#[macro_export] -macro_rules! unwrap_or_route { - ($result:expr, $route:expr) => { - match $result { - Ok(v) => v, - Err(_e) => { - return Ok($crate::TransformedFlowFile::route_without_changes($route)); - } - } +#[cfg(test)] +mod tests { + use super::*; + use crate::api::RawProcessor; + use crate::api::raw_processor::MultiThreadedTrigger; + use crate::{ + GetControllerService, GetId, MockFlowFile, MockLogger, MockProcessContext, + MockProcessSession, ProcessError, RouteErrorExt, }; - ($result:expr, $route:expr, $custom_logger:expr) => { - match $result { - Ok(v) => v, - Err(e) => { - $crate::error!( - $custom_logger, - "Failed to unwrap due to {}. Routing flow file.", - e - ); - return Ok($crate::TransformedFlowFile::route_without_changes($route)); - } + struct RouteToFailure; + impl Schedule for RouteToFailure { + fn schedule<Ctx: GetProperty, L: Logger>(_c: &Ctx, _l: &L) -> Result<Self, MinifiError> { + Ok(RouteToFailure) } - }; + } + impl FlowFileTransform for RouteToFailure { + fn transform< + 'a, + Context: GetProperty + GetControllerService + GetAttribute + GetId, + LoggerImpl: Logger, + >( + &self, + _context: &Context, + _input_stream: &'a mut dyn InputStream, + _logger: &LoggerImpl, + ) -> Result<TransformedFlowFile<'a>, ProcessError> { + let bad: Result<TransformedFlowFile<'a>, std::io::Error> = Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "bad data", + )); + bad.route_err_to_failure() + } + } - ($result:expr, $route:expr, $custom_logger:expr, $context:expr) => { - match $result { - Ok(v) => v, - Err(e) => { - $crate::error!( - $custom_logger, - "Failed to {} due to {}. Routing flow file.", - $context, - e - ); - return Ok($crate::TransformedFlowFile::route_without_changes($route)); - } + struct FatalTransform; + impl Schedule for FatalTransform { + fn schedule<Ctx: GetProperty, L: Logger>(_c: &Ctx, _l: &L) -> Result<Self, MinifiError> { + Ok(FatalTransform) } - }; + } + impl FlowFileTransform for FatalTransform { + fn transform< + 'a, + Context: GetProperty + GetControllerService + GetAttribute + GetId, + LoggerImpl: Logger, + >( + &self, + _context: &Context, + _input_stream: &'a mut dyn InputStream, + _logger: &LoggerImpl, + ) -> Result<TransformedFlowFile<'a>, ProcessError> { + Err(ProcessError::Fatal(MinifiError::custom("real error"))) + } + } + + fn seeded_session() -> MockProcessSession { + let mut session = MockProcessSession::new(); + session + .input_flow_files + .push(MockFlowFile::with_content(b"data")); + session + } + + #[test] + fn route_error_transfers_to_failure_and_commits() { + let mut processor: Processor< + RouteToFailure, + FlowFileTransformProcessorType, + MultiThreaded, + MockLogger, + > = Processor::new(MockLogger::new()); + processor.scheduled_impl = Some(RouteToFailure); + + let mut context = MockProcessContext::new(); + let mut session = seeded_session(); + + let result = MultiThreadedTrigger::trigger(&processor, &mut context, &mut session); + + assert_eq!( + result.expect("should commit, not roll back"), + OnTriggerResult::Ok + ); + let transferred = session.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 1); + assert_eq!(transferred[0].relationship, "failure"); + } + + #[test] + fn fatal_error_propagates_and_transfers_nothing() { + let mut processor: Processor< + FatalTransform, + FlowFileTransformProcessorType, + MultiThreaded, + MockLogger, + > = Processor::new(MockLogger::new()); + processor.scheduled_impl = Some(FatalTransform); + + let mut context = MockProcessContext::new(); + let mut session = seeded_session(); + + let result = MultiThreadedTrigger::trigger(&processor, &mut context, &mut session); + + assert!(matches!( + result, + Err(ProcessError::Fatal(MinifiError::CustomError(_))) + )); + assert_eq!(session.num_of_transferred_flow_files(), 0); + } } diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs index e22b3bab9..26b98128c 100644 --- a/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs +++ b/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs @@ -17,3 +17,4 @@ pub(crate) mod context_session_flowfile_bundle; pub(crate) mod flow_file_content; +pub(crate) mod with_attributes; diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs new file mode 100644 index 000000000..20e541c34 --- /dev/null +++ b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs @@ -0,0 +1,30 @@ +#[macro_export] +macro_rules! impl_with_attributes { + ($name:ident $(<$lt:lifetime>)?) => { + impl $(<$lt>)? $name $(<$lt>)? { + #[must_use] + pub fn with_attribute( + mut self, + key: impl Into<std::borrow::Cow<'static, str>>, + value: impl Into<std::borrow::Cow<'static, str>>, + ) -> Self { + self.attributes_to_add.push((key.into(), value.into()).into()); + self + } + + #[must_use] + pub fn with_attributes<K, V>( + mut self, + attributes: impl IntoIterator<Item = (K, V)> + ) -> Self + where + K: Into<std::borrow::Cow<'static, str>>, + V: Into<std::borrow::Cow<'static, str>>, + { + self.attributes_to_add + .extend(attributes.into_iter().map(|(k, v)| (k.into(), v.into()).into())); + self + } + } + }; +} diff --git a/minifi_rust/minifi_native/src/api/property.rs b/minifi_rust/minifi_native/src/api/property.rs index f541017ef..f8ff1368f 100644 --- a/minifi_rust/minifi_native/src/api/property.rs +++ b/minifi_rust/minifi_native/src/api/property.rs @@ -26,7 +26,7 @@ use std::marker::PhantomData; use std::str::FromStr; use std::time::Duration; -#[derive(Debug, Eq, PartialEq)] +#[derive(Debug, Eq, PartialEq, Clone)] pub enum StandardPropertyValidator { NonBlankValidator, TimePeriodValidator, @@ -38,13 +38,14 @@ pub enum StandardPropertyValidator { F64Validator, } -#[derive(Debug, PartialEq)] +#[derive(Debug, PartialEq, Clone)] pub enum PropertyConstraints { Validator(StandardPropertyValidator), AllowedValues(&'static [&'static str]), ControllerService(&'static str), } +#[derive(Clone)] pub struct PropertyDefinition { pub name: &'static str, pub description: &'static str, @@ -162,7 +163,7 @@ impl<T: PropertyType> PropertyValue for T { fn from_raw(raw: Option<String>, name: &str) -> Result<Self::Output, MinifiError> { match raw { Some(value) => T::parse(&value), - None => Err(MinifiError::missing_required_property(name.to_string())), + None => Err(MinifiError::MissingRequiredProperty(name.to_owned().into())), } } } @@ -286,7 +287,7 @@ where type Cs = Cs; type Output<'a> = &'a Cs; fn from_service<'a>(service: Option<&'a Cs>, name: &str) -> Result<&'a Cs, MinifiError> { - service.ok_or_else(|| MinifiError::missing_required_property(name.to_string())) + service.ok_or_else(|| MinifiError::MissingRequiredProperty(name.to_owned().into())) } } diff --git a/minifi_rust/minifi_native/src/api/raw_processor.rs b/minifi_rust/minifi_native/src/api/raw_processor.rs index 51a4a0550..9ae3c8ccc 100644 --- a/minifi_rust/minifi_native/src/api/raw_processor.rs +++ b/minifi_rust/minifi_native/src/api/raw_processor.rs @@ -16,7 +16,7 @@ // under the License. use crate::api::errors::MinifiError; -use crate::{LogLevel, Logger, ProcessContext, ProcessSession}; +use crate::{LogLevel, Logger, ProcessContext, ProcessError, ProcessSession}; pub enum ProcessorInputRequirement { Required, @@ -69,7 +69,7 @@ pub trait SingleThreadedTrigger: RawProcessor<Threading = SingleThreaded> { &mut self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>; @@ -80,7 +80,7 @@ pub trait MultiThreadedTrigger: RawProcessor<Threading = MultiThreaded> { &self, context: &mut PC, session: &mut PS, - ) -> Result<OnTriggerResult, MinifiError> + ) -> Result<OnTriggerResult, ProcessError> where PC: ProcessContext, PS: ProcessSession<FlowFile = PC::FlowFile>; diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs index 49df13f24..a2207ca85 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs @@ -517,11 +517,14 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { let mut reader = CffiInputStream::new(stream_ptr); if let Some(cb) = ctx.callback.take() { - match cb(&mut reader) { - Ok(cb_ok) => ctx.result = Some(Ok(cb_ok)), - Err(_) => { - return minifi_io_status_MINIFI_IO_ERROR; - } + let is_err = { + let outcome = cb(&mut reader); + let is_err = outcome.is_err(); + ctx.result = Some(outcome); + is_err + }; + if is_err { + return minifi_io_status_MINIFI_IO_ERROR; } } else { return minifi_io_status_MINIFI_IO_ERROR; @@ -538,6 +541,10 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { &mut ctx as *mut _ as *mut c_void, ); + if let Some(result) = ctx.result.take() { + return result; + } + if status != minifi_status_MINIFI_STATUS_SUCCESS { return Err(MinifiError::StatusError(( "minifi_process_session_read".into(), @@ -545,11 +552,7 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { ))); } - if let Some(result) = ctx.result.take() { - result - } else { - Err(MinifiError::UnknownError) - } + Err(MinifiError::UnknownError) } } diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs index a5c6688c0..c2b02407f 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs @@ -30,9 +30,37 @@ use crate::{ ComponentIdentifier, LogLevel, MultiThreaded, OutputAttribute, Processor, ProcessorDefinition, PropertyDefinition, Schedule, SingleThreaded, }; -use crate::{OnTriggerResult, Relationship}; +use crate::{OnTriggerResult, ProcessError, Relationship}; use minifi_native_sys::*; +fn process_error_to_status<P: RawProcessor>( + processor: &P, + result: Result<OnTriggerResult, ProcessError>, +) -> minifi_status { + match result { + Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS, + Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD, + Err(ProcessError::Fatal(err)) => { + processor.log( + LogLevel::Error, + format_args!("Error during trigger {}", err), + ); + err.to_status() + } + Err(ProcessError::Route(route)) => { + processor.log( + LogLevel::Warn, + format_args!( + "Cannot route to '{}' at the top level (no current flow file); \ + failing the trigger: {}", + route.relationship, route.source + ), + ); + minifi_status_MINIFI_STATUS_UNKNOWN_ERROR + } + } +} + pub trait DispatchOnTrigger<M: ThreadingModel> { /// # Safety /// @@ -57,17 +85,7 @@ where let processor = &*(processor_ptr as *const T); let mut context = CffiProcessContext::new(context_ptr); let mut session = CffiProcessSession::new(session_ptr); - match processor.trigger(&mut context, &mut session) { - Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS, - Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD, - Err(minifi_error) => { - processor.log( - LogLevel::Error, - format_args!("Error during trigger {}", minifi_error), - ); - minifi_error.to_status() - } - } + process_error_to_status(processor, processor.trigger(&mut context, &mut session)) } } } @@ -85,17 +103,8 @@ where let processor = &mut *(processor_ptr as *mut T); let mut context = CffiProcessContext::new(context_ptr); let mut session = CffiProcessSession::new(session_ptr); - match processor.trigger(&mut context, &mut session) { - Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS, - Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD, - Err(minifi_error) => { - processor.log( - LogLevel::Error, - format_args!("Error during trigger {}", minifi_error), - ); - minifi_error.to_status() - } - } + let result = processor.trigger(&mut context, &mut session); + process_error_to_status(&*processor, result) } } } @@ -284,7 +293,7 @@ where Implementation::SUPPORTS_DYNAMIC_RELATIONSHIPS, Implementation::OUTPUT_ATTRIBUTES, Implementation::RELATIONSHIPS, - Implementation::PROPERTIES, + Implementation::properties(), )) } } diff --git a/minifi_rust/minifi_native/src/lib.rs b/minifi_rust/minifi_native/src/lib.rs index ce54f9090..b855291a2 100644 --- a/minifi_rust/minifi_native/src/lib.rs +++ b/minifi_rust/minifi_native/src/lib.rs @@ -20,7 +20,7 @@ mod api; pub mod c_ffi; pub mod mock; -pub use api::errors::MinifiError; +pub use api::errors::{MinifiError, ProcessError, RouteError, RouteErrorExt}; pub use api::component_definition_traits::{ ComponentIdentifier, ControllerServiceDefinition, ProcessorDefinition, @@ -35,7 +35,7 @@ pub use api::processor_wrappers::flow_file_stream_transform::{ TransformStreamResult, }; pub use api::processor_wrappers::flow_file_transform::{ - FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile, + FlowFileAttribute, FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile, }; pub use api::processor_wrappers::utils::flow_file_content::Content;
