You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2026/08/13 14:11:37 UTC
(nifi-minifi-cpp) branch minifi_rust_impr_2 updated (ac288d9b4 -> 66ec2ddc0)
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a change to branch minifi_rust_impr_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
discard ac288d9b4 MINIFICPP-2886 Improve rust api process errors
new 66ec2ddc0 MINIFICPP-2886 Improve rust api process errors
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (ac288d9b4)
\
N -- N -- N refs/heads/minifi_rust_impr_2 (66ec2ddc0)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
minifi_rust/minifi_native/src/api/errors.rs | 7 -------
1 file changed, 7 deletions(-)
(nifi-minifi-cpp) 01/01: MINIFICPP-2886 Improve rust api process errors
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch minifi_rust_impr_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 66ec2ddc0878b4753d33a46cbc455e06a8d23a7a
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 | 36 ++--
.../src/processors/zoo_processor.rs | 14 +-
minifi_rust/minifi_native/src/api/attribute.rs | 11 +-
.../src/api/component_definition_traits.rs | 3 +-
minifi_rust/minifi_native/src/api/errors.rs | 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 | 222 ++++++++++++++------
.../src/api/processor_wrappers/utils.rs | 1 +
.../processor_wrappers/utils/with_attributes.rs | 30 +++
minifi_rust/minifi_native/src/api/property.rs | 9 +-
minifi_rust/minifi_native/src/api/raw_processor.rs | 6 +-
.../src/c_ffi/c_ffi_process_session.rs | 23 ++-
.../src/c_ffi/c_ffi_processor_definition.rs | 57 +++---
minifi_rust/minifi_native/src/lib.rs | 4 +-
minifi_rust/minifi_native_sys/build.rs | 3 +-
36 files changed, 666 insertions(+), 316 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 959ef7820..b5c9045e5 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file.rs
@@ -19,8 +19,8 @@
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
- GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule,
- Trigger,
+ GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, Schedule, Trigger,
};
use rand::RngExt;
use rand::distr::Alphanumeric;
@@ -147,7 +147,7 @@ impl Trigger for GenerateFlowFileRs {
context: &mut PC,
session: &mut PS,
_logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -164,7 +164,7 @@ impl Trigger for GenerateFlowFileRs {
context
.get_raw_property(&properties::CUSTOM_TEXT, None)?
.ok_or_else(|| {
- MinifiError::trigger_err(
+ MinifiError::custom(
"GenerateFlowFile is in CustomText mode but the \"Custom Text\" \
property is not set at trigger time",
)
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
index d3f9a9391..a3e0e8a68 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/generate_flow_file/processor_definition.rs
@@ -29,11 +29,14 @@ impl ProcessorDefinition for GenerateFlowFileRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- FILE_SIZE,
- BATCH_SIZE,
- DATA_FORMAT,
- UNIQUE_FLOW_FILES,
- CUSTOM_TEXT
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ FILE_SIZE,
+ BATCH_SIZE,
+ DATA_FORMAT,
+ UNIQUE_FLOW_FILES,
+ CUSTOM_TEXT
+ ];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
index ab2f0c037..7ec9f4130 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs
@@ -26,8 +26,8 @@ use crate::processors::get_file::properties::{
};
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
- GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession,
- Schedule, Trigger, debug, info, trace, warn,
+ GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, Schedule, Trigger, debug, info, trace, warn,
};
use std::collections::VecDeque;
use std::error;
@@ -225,7 +225,7 @@ impl Schedule for GetFileRs {
{
let input_directory = context.get_property(&DIRECTORY)?;
if !input_directory.is_dir() {
- return Err(MinifiError::schedule_err(format!(
+ return Err(MinifiError::custom(format!(
"{:?} is not a valid directory",
input_directory
)));
@@ -269,7 +269,7 @@ impl Trigger for GetFileRs {
context: &mut PC,
session: &mut PS,
logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
index 59171d3d0..13dbbbbdc 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/processor_definition.rs
@@ -33,16 +33,19 @@ impl ProcessorDefinition for GetFileRs {
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] =
&[ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- DIRECTORY,
- POLLING_INTERVAL,
- RECURSE,
- KEEP_SOURCE_FILE,
- MIN_AGE,
- MAX_AGE,
- MIN_SIZE,
- MAX_SIZE,
- IGNORE_HIDDEN_FILES,
- BATCH_SIZE,
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ DIRECTORY,
+ POLLING_INTERVAL,
+ RECURSE,
+ KEEP_SOURCE_FILE,
+ MIN_AGE,
+ MAX_AGE,
+ MIN_SIZE,
+ MAX_SIZE,
+ IGNORE_HIDDEN_FILES,
+ BATCH_SIZE,
+ ];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
index 50691c773..d7b9ed96f 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/get_file/tests.rs
@@ -38,10 +38,7 @@ fn schedule_fails_with_invalid_input_dir() {
"Input Directory".to_string(),
"/invalid_directory".to_string(),
);
- assert!(matches!(
- GetFileRs::schedule(&context, &MockLogger::new()),
- Err(MinifiError::ScheduleError(_))
- ));
+ assert!(GetFileRs::schedule(&context, &MockLogger::new()).is_err());
}
#[test]
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
index f6f843576..0faa04f18 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor.rs
@@ -25,8 +25,8 @@ use crate::processors::kamikaze_processor::properties::{
};
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
- GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule,
- Trigger,
+ GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, Schedule, Trigger,
};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -57,7 +57,7 @@ impl Schedule for KamikazeProcessorRs {
let schedule_behaviour = context.get_property(&SCHEDULE_BEHAVIOUR)?;
match schedule_behaviour {
- KamikazeBehaviour::ReturnErr => Err(MinifiError::schedule_err(
+ KamikazeBehaviour::ReturnErr => Err(MinifiError::custom(
"it was designed to fail during schedule",
)),
KamikazeBehaviour::ReturnOk => Ok(KamikazeProcessorRs { trigger_behaviour }),
@@ -81,16 +81,16 @@ impl Trigger for KamikazeProcessorRs {
context: &mut PC,
_session: &mut PS,
_logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
L: Logger,
{
match self.trigger_behaviour {
- KamikazeBehaviour::ReturnErr => Err(MinifiError::trigger_err(
- "it was designed to fail in trigger",
- )),
+ KamikazeBehaviour::ReturnErr => {
+ Err(MinifiError::custom("it was designed to fail in trigger").into())
+ }
KamikazeBehaviour::ReturnOk => Ok(OnTriggerResult::Ok),
KamikazeBehaviour::Panic => {
panic!("KamikazeProcessor::trigger panic")
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
index 51c8ff0c8..f9a47d115 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/processor_definition.rs
@@ -28,8 +28,9 @@ impl ProcessorDefinition for KamikazeProcessorRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- properties::SCHEDULE_BEHAVIOUR,
- properties::TRIGGER_BEHAVIOUR
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
index dd6e4794f..2b7f52166 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/kamikaze_processor/tests.rs
@@ -17,8 +17,7 @@
use super::*;
use crate::processors::kamikaze_processor::properties::{SCHEDULE_BEHAVIOUR, TRIGGER_BEHAVIOUR};
-use minifi_native::MinifiError::{ScheduleError, TriggerError};
-use minifi_native::{MockLogger, MockProcessContext, MockProcessSession};
+use minifi_native::{MockLogger, MockProcessContext, MockProcessSession, ProcessError};
use std::panic::AssertUnwindSafe;
#[test]
@@ -36,7 +35,7 @@ fn on_schedule_err() {
"ReturnErr".to_string(),
);
let processor = KamikazeProcessorRs::schedule(&context, &MockLogger::new());
- assert!(matches!(processor, Err(ScheduleError(_))));
+ assert!(matches!(processor, Err(MinifiError::CustomError(_))));
}
#[test]
@@ -78,7 +77,7 @@ fn on_trigger_err() {
let mut session = MockProcessSession::new();
assert!(matches!(
processor.trigger(&mut context, &mut session, &MockLogger::new()),
- Err(TriggerError(_))
+ Err(ProcessError::Fatal(MinifiError::CustomError(_)))
));
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
index 48de81360..329e281f4 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute.rs
@@ -21,8 +21,9 @@ use crate::processors::log_attribute::properties::{FLOW_FILES_TO_LOG, LOG_LEVEL,
use minifi_native::StandardPropertyValidator::NonBlankValidator;
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
- GetProperty, LogLevel, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession,
- PropertyConstraints, PropertySchema, PropertyType, Schedule, Trigger, debug, log, trace,
+ GetProperty, LogLevel, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessError,
+ ProcessSession, PropertyConstraints, PropertySchema, PropertyType, Schedule, Trigger, debug,
+ log, trace,
};
mod properties;
@@ -104,7 +105,7 @@ impl Trigger for LogAttributeRs {
_context: &mut PC,
session: &mut PS,
logger: &L,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
index b5732645d..91e0b3fed 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/log_attribute/processor_definition.rs
@@ -30,13 +30,16 @@ impl ProcessorDefinition for LogAttributeRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[relationships::SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- LOG_LEVEL,
- ATTRIBUTES_TO_LOG,
- ATTRIBUTES_TO_IGNORE,
- LOG_PAYLOAD,
- LOG_PREFIX,
- FLOW_FILES_TO_LOG,
- HEX_ENCODE_PAYLOAD,
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] = property_definitions![
+ LOG_LEVEL,
+ ATTRIBUTES_TO_LOG,
+ ATTRIBUTES_TO_IGNORE,
+ LOG_PAYLOAD,
+ LOG_PREFIX,
+ FLOW_FILES_TO_LOG,
+ HEX_ENCODE_PAYLOAD,
+ ];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
index f8a684d9f..4d32739e2 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user.rs
@@ -25,9 +25,8 @@ use crate::processors::lorem_ipsum_cs_user::relationships::SUCCESS;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
Content, FlowFileSource, GeneratedFlowFile, GetControllerService, GetProperty, Logger,
- MinifiError, Schedule, trace,
+ MinifiError, ProcessError, Schedule, trace,
};
-use std::collections::HashMap;
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
#[derive(
@@ -59,7 +58,7 @@ impl FlowFileSource for LoremIpsumCSUser {
&self,
context: &'a mut Context,
logger: &LoggerImpl,
- ) -> Result<Vec<GeneratedFlowFile<'a>>, MinifiError> {
+ ) -> Result<Vec<GeneratedFlowFile<'a>>, ProcessError> {
trace!(logger, "generate call {:?}", self);
let dummy_controller_service = context.get_controller_service(&DUMMY_CONTROLLER_SERVICE)?;
trace!(
@@ -72,15 +71,13 @@ impl FlowFileSource for LoremIpsumCSUser {
let generated_flow_file = GeneratedFlowFile::new(
&SUCCESS,
Some(Content::from(controller_service.data.clone())),
- HashMap::new(),
);
Ok(vec![generated_flow_file])
}
WriteMethod::Stream => {
let reader = controller_service.data.as_bytes();
let content = Content::Stream(Box::new(reader));
- let generated_flow_file =
- GeneratedFlowFile::new(&SUCCESS, Some(content), HashMap::new());
+ let generated_flow_file = GeneratedFlowFile::new(&SUCCESS, Some(content));
Ok(vec![generated_flow_file])
}
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
index 78c09ef0c..a80605035 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/lorem_ipsum_cs_user/processor_definition.rs
@@ -31,6 +31,9 @@ impl ProcessorDefinition for LoremIpsumCSUser {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS];
- const PROPERTIES: &'static [PropertyDefinition] =
- property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![CONTROLLER_SERVICE, DUMMY_CONTROLLER_SERVICE, WRITE_METHOD];
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
index e81c861ca..1dd63ec2d 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs
@@ -22,7 +22,7 @@ use crate::processors::put_file::unix_permissions::PutFileUnixPermissions;
use minifi_native::macros::{ComponentIdentifier, PropertyType};
use minifi_native::{
FlowFileTransform, GetAttribute, GetControllerService, GetId, GetProperty, InputStream, Logger,
- MinifiError, Schedule, TransformedFlowFile, trace, unwrap_or_route, warn,
+ MinifiError, ProcessError, RouteErrorExt, Schedule, TransformedFlowFile, trace, warn,
};
use std::path::{Path, PathBuf};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -171,11 +171,10 @@ impl FlowFileTransform for PutFileRs {
context: &Context,
input_stream: &'a mut dyn InputStream,
logger: &LoggerImpl,
- ) -> Result<TransformedFlowFile<'a>, MinifiError> {
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
trace!(logger, "on_trigger: {:?}", self);
- let destination_path =
- unwrap_or_route!(Self::get_destination_path(context), &FAILURE, logger);
+ let destination_path = Self::get_destination_path(context).route_err_to_failure()?;
if self.directory_is_full(&destination_path) {
warn!(logger, "Directory is full");
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
index 93be840a4..9b63c3269 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file/processor_definition.rs
@@ -30,21 +30,25 @@ impl ProcessorDefinition for PutFileRs {
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[SUCCESS, FAILURE];
- #[cfg(windows)]
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- properties::DIRECTORY,
- properties::CONFLICT_RESOLUTION,
- properties::CREATE_DIRS,
- properties::MAX_FILE_COUNT,
- ];
-
#[cfg(unix)]
- const PROPERTIES: &'static [PropertyDefinition] = property_definitions![
- properties::DIRECTORY,
- properties::CONFLICT_RESOLUTION,
- properties::CREATE_DIRS,
- properties::MAX_FILE_COUNT,
- unix_only_properties::PERMISSIONS,
- unix_only_properties::DIRECTORY_PERMISSIONS,
- ];
+ fn properties() -> &'static [PropertyDefinition] {
+ use std::sync::LazyLock;
+ static COMBINED_PROPERTIES: LazyLock<Vec<PropertyDefinition>> = LazyLock::new(|| {
+ let mut props = Vec::new();
+ props.extend_from_slice(property_definitions![
+ properties::DIRECTORY,
+ properties::CONFLICT_RESOLUTION,
+ properties::CREATE_DIRS,
+ properties::MAX_FILE_COUNT,
+ ]);
+ #[cfg(unix)]
+ {
+ props.push(unix_only_properties::PERMISSIONS.definition());
+ props.push(unix_only_properties::DIRECTORY_PERMISSIONS.definition());
+ }
+ props
+ });
+
+ &COMBINED_PROPERTIES
+ }
}
diff --git a/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
index 86d67c779..6eabf7a4d 100644
--- a/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
+++ b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
@@ -21,8 +21,8 @@ use crate::controller_services::animal_controller_apis::{
use minifi_native::macros::ComponentIdentifier;
use minifi_native::{
GetProperty, Logger, MinifiError, OnTriggerResult, OutputAttribute, ProcessContext,
- ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property, PropertyDefinition,
- Relationship, Schedule, Trigger, critical, info, property_definitions,
+ ProcessError, ProcessSession, ProcessorDefinition, ProcessorInputRequirement, Property,
+ PropertyDefinition, Relationship, Schedule, Trigger, critical, info, property_definitions,
};
pub(crate) const CAN_FLY_SERVICE: Property<dyn CanFlyControllerApi> =
@@ -52,7 +52,7 @@ impl Trigger for ZooProcessorRs {
context: &mut Context,
_session: &mut Session,
logger: &Lggr,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
Context: ProcessContext,
Session: ProcessSession<FlowFile = Context::FlowFile>,
@@ -81,6 +81,10 @@ impl ProcessorDefinition for ZooProcessorRs {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[];
const RELATIONSHIPS: &'static [Relationship] = &[];
- const PROPERTIES: &'static [PropertyDefinition] =
- property_definitions![CAN_FLY_SERVICE, NUMBER_OF_LEGS];
+ fn properties() -> &'static [PropertyDefinition] {
+ const PROPERTIES: &[PropertyDefinition] =
+ property_definitions![CAN_FLY_SERVICE, NUMBER_OF_LEGS];
+
+ PROPERTIES
+ }
}
diff --git a/minifi_rust/minifi_native/src/api/attribute.rs b/minifi_rust/minifi_native/src/api/attribute.rs
index 571ea7ce7..ed0301654 100644
--- a/minifi_rust/minifi_native/src/api/attribute.rs
+++ b/minifi_rust/minifi_native/src/api/attribute.rs
@@ -16,6 +16,7 @@
// under the License.
use crate::MinifiError;
+use std::borrow::Cow;
pub struct OutputAttribute {
pub name: &'static str,
@@ -23,10 +24,18 @@ pub struct OutputAttribute {
pub description: &'static str,
}
+impl From<&OutputAttribute> for Cow<'static, str> {
+ fn from(attr: &OutputAttribute) -> Self {
+ Cow::Borrowed(attr.name)
+ }
+}
+
pub trait GetAttribute {
fn get_attribute(&self, name: &str) -> Result<Option<String>, MinifiError>;
fn get_required_attribute(&self, name: &str) -> Result<String, MinifiError> {
self.get_attribute(name)?
- .ok_or(MinifiError::missing_required_attribute(name.to_owned()))
+ .ok_or(MinifiError::MissingRequiredAttribute(
+ name.to_owned().into(),
+ ))
}
}
diff --git a/minifi_rust/minifi_native/src/api/component_definition_traits.rs b/minifi_rust/minifi_native/src/api/component_definition_traits.rs
index 5e0e8a126..c1f473f42 100644
--- a/minifi_rust/minifi_native/src/api/component_definition_traits.rs
+++ b/minifi_rust/minifi_native/src/api/component_definition_traits.rs
@@ -31,7 +31,8 @@ pub trait ProcessorDefinition {
const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool;
const OUTPUT_ATTRIBUTES: &'static [OutputAttribute];
const RELATIONSHIPS: &'static [Relationship];
- const PROPERTIES: &'static [PropertyDefinition];
+ /// fn to allow concatenation for complex processors
+ fn properties() -> &'static [PropertyDefinition];
}
pub trait ControllerServiceDefinition: Sized + 'static {
diff --git a/minifi_rust/minifi_native/src/api/errors.rs b/minifi_rust/minifi_native/src/api/errors.rs
index 2c833a997..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 ef2342b5b..462b70973 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use crate::api::InputStream;
use crate::api::flow_file::GetId;
use crate::api::processor::Processor;
use crate::api::processor_wrappers::utils::context_session_flowfile_bundle::ContextSessionFlowFileBundle;
@@ -24,50 +23,68 @@ use crate::api::property::{GetControllerService, GetProperty};
use crate::api::raw_processor::{MultiThreadedTrigger, SingleThreadedTrigger};
use crate::{
GetAttribute, LogLevel, Logger, MinifiError, MultiThreaded, OnTriggerResult, ProcessContext,
- ProcessSession, Relationship, Schedule, SingleThreaded, info,
+ ProcessError, ProcessSession, Relationship, Schedule, SingleThreaded, impl_with_attributes,
+ info,
};
-use std::collections::HashMap;
+
+use minifi_native::InputStream;
+use std::borrow::Cow;
+
+pub type FlowFileAttribute = (Cow<'static, str>, Cow<'static, str>);
#[derive(Debug)]
pub struct TransformedFlowFile<'a> {
- target_relationship_name: &'static str,
+ target_relationship_name: Cow<'static, str>,
new_content: Option<Content<'a>>, // If None, the content doesn't change
- attributes_to_add: HashMap<String, String>,
+ attributes_to_add: Vec<FlowFileAttribute>,
}
impl<'a> TransformedFlowFile<'a> {
pub fn route_without_changes(target_relationship: &Relationship) -> Self {
+ Self::route_without_changes_by_name(Cow::Borrowed(target_relationship.name))
+ }
+
+ pub fn route_without_changes_by_name(relationship: Cow<'static, str>) -> Self {
Self {
- target_relationship_name: target_relationship.name,
+ target_relationship_name: relationship,
new_content: None,
- attributes_to_add: HashMap::new(),
+ attributes_to_add: Vec::new(),
}
}
- pub fn new(
- target_relationship: &Relationship,
- new_content: Option<Vec<u8>>,
- attributes_to_add: HashMap<String, String>,
- ) -> Self {
+ pub fn new(target_relationship: &Relationship, new_content: Option<Vec<u8>>) -> Self {
Self {
- target_relationship_name: target_relationship.name,
+ target_relationship_name: Cow::Borrowed(target_relationship.name),
new_content: new_content.map(Content::Buffer),
- attributes_to_add,
+ attributes_to_add: Vec::new(),
}
}
+ #[must_use]
+ pub fn with_content(mut self, content: Vec<u8>) -> Self {
+ self.new_content = Some(Content::Buffer(content));
+ self
+ }
+
pub fn new_content(&'_ self) -> Option<&'_ Content<'_>> {
self.new_content.as_ref()
}
- pub fn target_relationship(&self) -> &'static str {
- self.target_relationship_name
+ pub fn target_relationship(&self) -> &str {
+ &self.target_relationship_name
}
- pub fn attributes_to_add(&self) -> &HashMap<String, String> {
+ pub fn attributes_to_add(&self) -> &[FlowFileAttribute] {
&self.attributes_to_add
}
+ pub fn attribute(&self, name: &str) -> Option<&str> {
+ self.attributes_to_add
+ .iter()
+ .find(|(k, _)| k == name)
+ .map(|(_, v)| v.as_ref())
+ }
+
#[cfg(any(test, feature = "test-utils"))]
pub fn into_bytes(self) -> std::io::Result<Option<Vec<u8>>> {
match self.new_content {
@@ -82,6 +99,8 @@ impl<'a> TransformedFlowFile<'a> {
}
}
+impl_with_attributes!(TransformedFlowFile<'a>);
+
pub trait FlowFileTransform {
fn transform<
'a,
@@ -92,7 +111,7 @@ pub trait FlowFileTransform {
context: &Context,
input_stream: &'a mut dyn InputStream,
logger: &LoggerImpl,
- ) -> Result<TransformedFlowFile<'a>, MinifiError>;
+ ) -> Result<TransformedFlowFile<'a>, ProcessError>;
}
pub trait MutFlowFileTransform {
@@ -105,7 +124,7 @@ pub trait MutFlowFileTransform {
context: &Context,
input_stream: &'a mut dyn InputStream,
logger: &LoggerImpl,
- ) -> Result<TransformedFlowFile<'a>, MinifiError>;
+ ) -> Result<TransformedFlowFile<'a>, ProcessError>;
}
pub struct FlowFileTransformProcessorType {}
@@ -115,7 +134,7 @@ fn handle_transform<PC, PS, L, F>(
session: &mut PS,
logger: &L,
mut transform_fn: F,
-) -> Result<OnTriggerResult, MinifiError>
+) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -123,13 +142,22 @@ where
F: for<'stream> FnMut(
&ContextSessionFlowFileBundle<'_, PC, PS>,
&'stream mut dyn InputStream,
- ) -> Result<TransformedFlowFile<'stream>, MinifiError>,
+ ) -> Result<TransformedFlowFile<'stream>, ProcessError>,
{
if let Some(mut flow_file) = session.get() {
let simple_context = ContextSessionFlowFileBundle::new(context, session, Some(&flow_file));
let (attrs_to_add, relationship) = session.read_stream(&flow_file, |input_stream| {
- let transformed = transform_fn(&simple_context, input_stream)?;
+ let transformed = match transform_fn(&simple_context, input_stream) {
+ Ok(transform_success) => transform_success,
+ Err(ProcessError::Route(route)) => {
+ route.log(logger);
+ TransformedFlowFile::route_without_changes_by_name(route.relationship)
+ }
+ Err(ProcessError::Fatal(e)) => {
+ return Err(e);
+ }
+ };
info!(logger, "{:?}", transformed);
match transformed.new_content {
@@ -152,7 +180,7 @@ where
session.set_attribute(&mut flow_file, &k, &v)?;
}
- session.transfer(flow_file, relationship)?;
+ session.transfer(flow_file, relationship.as_ref())?;
Ok(OnTriggerResult::Ok)
} else {
logger.log(LogLevel::Trace, format_args!("No flowfile to transform"));
@@ -170,7 +198,7 @@ where
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -180,9 +208,7 @@ where
scheduled_impl.transform(ctx, input, &self.logger)
})
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
@@ -197,7 +223,7 @@ where
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>,
@@ -207,48 +233,118 @@ where
scheduled_impl.transform(ctx, input, &self.logger)
})
} else {
- Err(MinifiError::trigger_err(
- "The processor hasn't been scheduled yet",
- ))
+ Err(MinifiError::UnscheduledProcessor.into())
}
}
}
-#[macro_export]
-macro_rules! unwrap_or_route {
- ($result:expr, $route:expr) => {
- match $result {
- Ok(v) => v,
- Err(_e) => {
- return Ok(TransformedFlowFile::route_without_changes($route));
- }
- }
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::api::RawProcessor;
+ use crate::api::raw_processor::MultiThreadedTrigger;
+ use crate::{
+ GetControllerService, GetId, MockFlowFile, MockLogger, MockProcessContext,
+ MockProcessSession, ProcessError, RouteErrorExt,
};
- ($result:expr, $route:expr, $custom_logger:expr) => {
- match $result {
- Ok(v) => v,
- Err(e) => {
- minifi_native::error!(
- $custom_logger,
- "Failed to unwrap due to {}. Routing flow file.",
- e
- );
- return Ok(TransformedFlowFile::route_without_changes($route));
- }
+ struct RouteToFailure;
+ impl Schedule for RouteToFailure {
+ fn schedule<Ctx: GetProperty, L: Logger>(_c: &Ctx, _l: &L) -> Result<Self, MinifiError> {
+ Ok(RouteToFailure)
}
- };
+ }
+ impl FlowFileTransform for RouteToFailure {
+ fn transform<
+ 'a,
+ Context: GetProperty + GetControllerService + GetAttribute + GetId,
+ LoggerImpl: Logger,
+ >(
+ &self,
+ _context: &Context,
+ _input_stream: &'a mut dyn InputStream,
+ _logger: &LoggerImpl,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ let bad: Result<TransformedFlowFile<'a>, std::io::Error> = Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "bad data",
+ ));
+ bad.route_err_to_failure()
+ }
+ }
- ($result:expr, $route:expr, $custom_logger:expr, $context:expr) => {
- match $result {
- Ok(v) => v,
- Err(e) => {
- error!(
- $custom_logger,
- "Failed to {} due to {}. Routing flow file.", $context, e
- );
- return Ok(TransformedFlowFile::route_without_changes($route));
- }
+ struct FatalTransform;
+ impl Schedule for FatalTransform {
+ fn schedule<Ctx: GetProperty, L: Logger>(_c: &Ctx, _l: &L) -> Result<Self, MinifiError> {
+ Ok(FatalTransform)
}
- };
+ }
+ impl FlowFileTransform for FatalTransform {
+ fn transform<
+ 'a,
+ Context: GetProperty + GetControllerService + GetAttribute + GetId,
+ LoggerImpl: Logger,
+ >(
+ &self,
+ _context: &Context,
+ _input_stream: &'a mut dyn InputStream,
+ _logger: &LoggerImpl,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ Err(ProcessError::Fatal(MinifiError::custom("real error")))
+ }
+ }
+
+ fn seeded_session() -> MockProcessSession {
+ let mut session = MockProcessSession::new();
+ session
+ .input_flow_files
+ .push(MockFlowFile::with_content(b"data"));
+ session
+ }
+
+ #[test]
+ fn route_error_transfers_to_failure_and_commits() {
+ let mut processor: Processor<
+ RouteToFailure,
+ FlowFileTransformProcessorType,
+ MultiThreaded,
+ MockLogger,
+ > = Processor::new(MockLogger::new());
+ processor.scheduled_impl = Some(RouteToFailure);
+
+ let mut context = MockProcessContext::new();
+ let mut session = seeded_session();
+
+ let result = MultiThreadedTrigger::trigger(&processor, &mut context, &mut session);
+
+ assert_eq!(
+ result.expect("should commit, not roll back"),
+ OnTriggerResult::Ok
+ );
+ let transferred = session.transferred_flow_files.borrow();
+ assert_eq!(transferred.len(), 1);
+ assert_eq!(transferred[0].relationship, "failure");
+ }
+
+ #[test]
+ fn fatal_error_propagates_and_transfers_nothing() {
+ let mut processor: Processor<
+ FatalTransform,
+ FlowFileTransformProcessorType,
+ MultiThreaded,
+ MockLogger,
+ > = Processor::new(MockLogger::new());
+ processor.scheduled_impl = Some(FatalTransform);
+
+ let mut context = MockProcessContext::new();
+ let mut session = seeded_session();
+
+ let result = MultiThreadedTrigger::trigger(&processor, &mut context, &mut session);
+
+ assert!(matches!(
+ result,
+ Err(ProcessError::Fatal(MinifiError::CustomError(_)))
+ ));
+ assert_eq!(session.num_of_transferred_flow_files(), 0);
+ }
}
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs
index e22b3bab9..26b98128c 100644
--- a/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/utils.rs
@@ -17,3 +17,4 @@
pub(crate) mod context_session_flowfile_bundle;
pub(crate) mod flow_file_content;
+pub(crate) mod with_attributes;
diff --git a/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs
new file mode 100644
index 000000000..20e541c34
--- /dev/null
+++ b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs
@@ -0,0 +1,30 @@
+#[macro_export]
+macro_rules! impl_with_attributes {
+ ($name:ident $(<$lt:lifetime>)?) => {
+ impl $(<$lt>)? $name $(<$lt>)? {
+ #[must_use]
+ pub fn with_attribute(
+ mut self,
+ key: impl Into<std::borrow::Cow<'static, str>>,
+ value: impl Into<std::borrow::Cow<'static, str>>,
+ ) -> Self {
+ self.attributes_to_add.push((key.into(), value.into()).into());
+ self
+ }
+
+ #[must_use]
+ pub fn with_attributes<K, V>(
+ mut self,
+ attributes: impl IntoIterator<Item = (K, V)>
+ ) -> Self
+ where
+ K: Into<std::borrow::Cow<'static, str>>,
+ V: Into<std::borrow::Cow<'static, str>>,
+ {
+ self.attributes_to_add
+ .extend(attributes.into_iter().map(|(k, v)| (k.into(), v.into()).into()));
+ self
+ }
+ }
+ };
+}
diff --git a/minifi_rust/minifi_native/src/api/property.rs b/minifi_rust/minifi_native/src/api/property.rs
index f0c95767f..f9fc011fc 100644
--- a/minifi_rust/minifi_native/src/api/property.rs
+++ b/minifi_rust/minifi_native/src/api/property.rs
@@ -26,7 +26,7 @@ use std::marker::PhantomData;
use std::str::FromStr;
use std::time::Duration;
-#[derive(Debug, Eq, PartialEq)]
+#[derive(Debug, Eq, PartialEq, Clone)]
pub enum StandardPropertyValidator {
NonBlankValidator,
TimePeriodValidator,
@@ -38,13 +38,14 @@ pub enum StandardPropertyValidator {
F64Validator,
}
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Clone)]
pub enum PropertyConstraints {
Validator(StandardPropertyValidator),
AllowedValues(&'static [&'static str]),
ControllerService(&'static str),
}
+#[derive(Clone)]
pub struct PropertyDefinition {
pub name: &'static str,
pub description: &'static str,
@@ -151,7 +152,7 @@ impl<T: PropertyType> PropertyValue for T {
fn from_raw(raw: Option<String>, name: &str) -> Result<Self::Output, MinifiError> {
match raw {
Some(value) => T::parse(&value),
- None => Err(MinifiError::missing_required_property(name.to_string())),
+ None => Err(MinifiError::MissingRequiredProperty(name.to_owned().into())),
}
}
}
@@ -274,7 +275,7 @@ where
type Cs = Cs;
type Output<'a> = &'a Cs;
fn from_service<'a>(service: Option<&'a Cs>, name: &str) -> Result<&'a Cs, MinifiError> {
- service.ok_or_else(|| MinifiError::missing_required_property(name.to_string()))
+ service.ok_or_else(|| MinifiError::MissingRequiredProperty(name.to_owned().into()))
}
}
diff --git a/minifi_rust/minifi_native/src/api/raw_processor.rs b/minifi_rust/minifi_native/src/api/raw_processor.rs
index 51a4a0550..9ae3c8ccc 100644
--- a/minifi_rust/minifi_native/src/api/raw_processor.rs
+++ b/minifi_rust/minifi_native/src/api/raw_processor.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::api::errors::MinifiError;
-use crate::{LogLevel, Logger, ProcessContext, ProcessSession};
+use crate::{LogLevel, Logger, ProcessContext, ProcessError, ProcessSession};
pub enum ProcessorInputRequirement {
Required,
@@ -69,7 +69,7 @@ pub trait SingleThreadedTrigger: RawProcessor<Threading = SingleThreaded> {
&mut self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>;
@@ -80,7 +80,7 @@ pub trait MultiThreadedTrigger: RawProcessor<Threading = MultiThreaded> {
&self,
context: &mut PC,
session: &mut PS,
- ) -> Result<OnTriggerResult, MinifiError>
+ ) -> Result<OnTriggerResult, ProcessError>
where
PC: ProcessContext,
PS: ProcessSession<FlowFile = PC::FlowFile>;
diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs
index 49df13f24..a2207ca85 100644
--- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs
+++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs
@@ -517,11 +517,14 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
let mut reader = CffiInputStream::new(stream_ptr);
if let Some(cb) = ctx.callback.take() {
- match cb(&mut reader) {
- Ok(cb_ok) => ctx.result = Some(Ok(cb_ok)),
- Err(_) => {
- return minifi_io_status_MINIFI_IO_ERROR;
- }
+ let is_err = {
+ let outcome = cb(&mut reader);
+ let is_err = outcome.is_err();
+ ctx.result = Some(outcome);
+ is_err
+ };
+ if is_err {
+ return minifi_io_status_MINIFI_IO_ERROR;
}
} else {
return minifi_io_status_MINIFI_IO_ERROR;
@@ -538,6 +541,10 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
&mut ctx as *mut _ as *mut c_void,
);
+ if let Some(result) = ctx.result.take() {
+ return result;
+ }
+
if status != minifi_status_MINIFI_STATUS_SUCCESS {
return Err(MinifiError::StatusError((
"minifi_process_session_read".into(),
@@ -545,11 +552,7 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
)));
}
- if let Some(result) = ctx.result.take() {
- result
- } else {
- Err(MinifiError::UnknownError)
- }
+ Err(MinifiError::UnknownError)
}
}
diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs
index a5c6688c0..c2b02407f 100644
--- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs
+++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_processor_definition.rs
@@ -30,9 +30,37 @@ use crate::{
ComponentIdentifier, LogLevel, MultiThreaded, OutputAttribute, Processor, ProcessorDefinition,
PropertyDefinition, Schedule, SingleThreaded,
};
-use crate::{OnTriggerResult, Relationship};
+use crate::{OnTriggerResult, ProcessError, Relationship};
use minifi_native_sys::*;
+fn process_error_to_status<P: RawProcessor>(
+ processor: &P,
+ result: Result<OnTriggerResult, ProcessError>,
+) -> minifi_status {
+ match result {
+ Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
+ Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
+ Err(ProcessError::Fatal(err)) => {
+ processor.log(
+ LogLevel::Error,
+ format_args!("Error during trigger {}", err),
+ );
+ err.to_status()
+ }
+ Err(ProcessError::Route(route)) => {
+ processor.log(
+ LogLevel::Warn,
+ format_args!(
+ "Cannot route to '{}' at the top level (no current flow file); \
+ failing the trigger: {}",
+ route.relationship, route.source
+ ),
+ );
+ minifi_status_MINIFI_STATUS_UNKNOWN_ERROR
+ }
+ }
+}
+
pub trait DispatchOnTrigger<M: ThreadingModel> {
/// # Safety
///
@@ -57,17 +85,7 @@ where
let processor = &*(processor_ptr as *const T);
let mut context = CffiProcessContext::new(context_ptr);
let mut session = CffiProcessSession::new(session_ptr);
- match processor.trigger(&mut context, &mut session) {
- Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
- Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
- Err(minifi_error) => {
- processor.log(
- LogLevel::Error,
- format_args!("Error during trigger {}", minifi_error),
- );
- minifi_error.to_status()
- }
- }
+ process_error_to_status(processor, processor.trigger(&mut context, &mut session))
}
}
}
@@ -85,17 +103,8 @@ where
let processor = &mut *(processor_ptr as *mut T);
let mut context = CffiProcessContext::new(context_ptr);
let mut session = CffiProcessSession::new(session_ptr);
- match processor.trigger(&mut context, &mut session) {
- Ok(OnTriggerResult::Ok) => minifi_status_MINIFI_STATUS_SUCCESS,
- Ok(OnTriggerResult::Yield) => minifi_status_MINIFI_STATUS_PROCESSOR_YIELD,
- Err(minifi_error) => {
- processor.log(
- LogLevel::Error,
- format_args!("Error during trigger {}", minifi_error),
- );
- minifi_error.to_status()
- }
- }
+ let result = processor.trigger(&mut context, &mut session);
+ process_error_to_status(&*processor, result)
}
}
}
@@ -284,7 +293,7 @@ where
Implementation::SUPPORTS_DYNAMIC_RELATIONSHIPS,
Implementation::OUTPUT_ATTRIBUTES,
Implementation::RELATIONSHIPS,
- Implementation::PROPERTIES,
+ Implementation::properties(),
))
}
}
diff --git a/minifi_rust/minifi_native/src/lib.rs b/minifi_rust/minifi_native/src/lib.rs
index ce54f9090..b855291a2 100644
--- a/minifi_rust/minifi_native/src/lib.rs
+++ b/minifi_rust/minifi_native/src/lib.rs
@@ -20,7 +20,7 @@ mod api;
pub mod c_ffi;
pub mod mock;
-pub use api::errors::MinifiError;
+pub use api::errors::{MinifiError, ProcessError, RouteError, RouteErrorExt};
pub use api::component_definition_traits::{
ComponentIdentifier, ControllerServiceDefinition, ProcessorDefinition,
@@ -35,7 +35,7 @@ pub use api::processor_wrappers::flow_file_stream_transform::{
TransformStreamResult,
};
pub use api::processor_wrappers::flow_file_transform::{
- FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile,
+ FlowFileAttribute, FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile,
};
pub use api::processor_wrappers::utils::flow_file_content::Content;
diff --git a/minifi_rust/minifi_native_sys/build.rs b/minifi_rust/minifi_native_sys/build.rs
index 3403c4bbe..1906478ad 100644
--- a/minifi_rust/minifi_native_sys/build.rs
+++ b/minifi_rust/minifi_native_sys/build.rs
@@ -228,7 +228,8 @@ fn main() {
let bindings = bindgen::Builder::default()
.header(sdk.header_path.to_str().unwrap())
- .clang_arg("-std=c2x")
+ .clang_arg("-include")
+ .clang_arg("stdbool.h")
.parse_callbacks(Box::new(bindgen::CargoCallbacks::new()))
.generate()
.expect("Unable to generate bindings");