You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2026/08/26 11:34:08 UTC
(nifi-minifi-cpp) 15/22: MINIFICPP-2886 Improve rust api process errors
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 <ma...@apache.org>
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;