This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 2559f40d2 MINIFICPP-2886 Improve rust api process errors (#2246)
2559f40d2 is described below
commit 2559f40d27d22fa58aaf2b18921ce0f9a1d7c776
Author: Martin Zink <[email protected]>
AuthorDate: Mon Sep 14 13:50:56 2026 +0200
MINIFICPP-2886 Improve rust api process errors (#2246)
I've added a new layer that helps with routing, the process error now
contains fatal and route errors.
One should use it based on the nature of the error. If its fatal the agent
will handle the issue, if its routing error that's technically not an error in
the eyes of the agent, we log it route to proper relationship then continue
with data processing.
I've changed the errors a bit (schedule, trigger with string literal ->
custom) it will already include the errors origin so no need to separate them
The TransformedFlowFile/GeneratedFlowFile has also been changed so its
easier to modify them after creation (with chained operators like
with_attribute(s)/with_content)
Additional squashed commits:
* add anyhow error support and fold parse errors into validation errors
* review changes
* revert const PROPERTIES -> const fn properties()
* tests for with_attribute(s)
* wildcard for generate_docs
* apache license for with_attributes.rs
* with_content changes
* build fix
* adds comment to with_attributes.rs
* cargo fmt
* simplify Cow relationships
---
.../features/error-handling.feature | 4 +-
.../src/processors/asciify_german.rs | 16 +-
.../src/processors/asciify_german/tests.rs | 10 +-
.../src/processors/count_actual_logging.rs | 6 +-
.../src/processors/duplicate_text.rs | 7 +-
.../src/processors/generate_flow_file.rs | 8 +-
.../src/processors/get_file.rs | 8 +-
.../src/processors/get_file/tests.rs | 5 +-
.../src/processors/kamikaze_processor.rs | 14 +-
.../src/processors/kamikaze_processor/tests.rs | 7 +-
.../src/processors/log_attribute.rs | 7 +-
.../src/processors/lorem_ipsum_cs_user.rs | 9 +-
.../src/processors/put_file.rs | 7 +-
.../src/processors/zoo_processor.rs | 6 +-
minifi_rust/generate_docs/Dockerfile.generate_docs | 2 +-
minifi_rust/minifi_native/Cargo.toml | 1 +
minifi_rust/minifi_native/src/api/attribute.rs | 11 +-
minifi_rust/minifi_native/src/api/errors.rs | 287 +++++++++++++++------
.../api/processor_wrappers/complex_processor.rs | 18 +-
.../src/api/processor_wrappers/flow_file_source.rs | 64 +++--
.../flow_file_stream_transform.rs | 93 +++++--
.../api/processor_wrappers/flow_file_transform.rs | 241 ++++++++++++-----
.../src/api/processor_wrappers/utils.rs | 1 +
.../processor_wrappers/utils/with_attributes.rs | 50 ++++
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 | 19 +-
.../src/c_ffi/c_ffi_processor_definition.rs | 55 ++--
minifi_rust/minifi_native/src/lib.rs | 4 +-
29 files changed, 674 insertions(+), 301 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/tests.rs
b/minifi_rust/extensions/minifi_rs_playground/src/processors/asciify_german/tests.rs
index c9b93c658..52863ac6c 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,9 @@ 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);
+ match result {
+ Err(ProcessError::Route(route)) => assert_eq!(route.relationship,
FAILURE.name),
+ other => panic!("expected a route error to failure, got {other:?}"),
+ }
}
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..f17d6734d 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>,
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..8f32de446 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))
}
}
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/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/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/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/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/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/zoo_processor.rs
b/minifi_rust/extensions/minifi_rs_playground/src/processors/zoo_processor.rs
index 86d67c779..90c063eed 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>,
diff --git a/minifi_rust/generate_docs/Dockerfile.generate_docs
b/minifi_rust/generate_docs/Dockerfile.generate_docs
index 3d0751617..6ecee62cb 100644
--- a/minifi_rust/generate_docs/Dockerfile.generate_docs
+++ b/minifi_rust/generate_docs/Dockerfile.generate_docs
@@ -20,7 +20,7 @@ ARG BASE_IMAGE="apacheminificpp:behave"
FROM ${BASE_IMAGE} AS builder
LABEL maintainer="Martin Zink <[email protected]>"
-COPY ./target/release/libminifi_rs_playground.so
/opt/minifi/minifi-current/extensions
+COPY ./target/release/libminifi_*.so /opt/minifi/minifi-current/extensions
RUN /opt/minifi/minifi-current/bin/minifi --docs DOCS_OUTPUT
diff --git a/minifi_rust/minifi_native/Cargo.toml
b/minifi_rust/minifi_native/Cargo.toml
index b3f75648b..8d0a74da2 100644
--- a/minifi_rust/minifi_native/Cargo.toml
+++ b/minifi_rust/minifi_native/Cargo.toml
@@ -13,6 +13,7 @@ strum_macros = "0.28.0"
humantime = "2.3.0"
byte-unit = "5.1.6"
itertools = "0.14.0"
+anyhow = "1.0.104"
[features]
test-utils = []
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/errors.rs
b/minifi_rust/minifi_native/src/api/errors.rs
index 2c833a997..16c652a7c 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;
@@ -23,87 +24,176 @@ use std::fmt;
use std::num::{NonZeroU32, ParseFloatError, ParseIntError};
use std::str::ParseBoolError;
-#[derive(Debug, Clone)]
-pub enum ParseError {
- Strum(strum::ParseError),
- Bool(ParseBoolError),
- Int(ParseIntError),
- Duration(humantime::DurationError),
- Size(byte_unit::ParseError),
- Nul(NulError),
- Float(ParseFloatError),
- Other,
-}
-
#[derive(Debug)]
-pub enum MinifiError {
- UnknownError,
- StatusError((Cow<'static, str>, NonZeroU32)),
- MissingRequiredAttribute(Cow<'static, str>),
- MissingRequiredProperty(Cow<'static, str>),
- ControllerServiceError(Cow<'static, str>),
- ValidationError(Cow<'static, str>),
- ScheduleError(Cow<'static, str>),
- TriggerError(Cow<'static, str>),
- Parse(ParseError),
- MissingFlowFileError,
- IoError(std::io::Error),
+pub struct RouteError {
+ pub relationship: &'static str,
+ pub source: Box<dyn Error + Send + Sync + 'static>,
+ pub log_level: LogLevel,
}
-impl From<std::io::Error> for MinifiError {
- fn from(error: std::io::Error) -> Self {
- MinifiError::IoError(error)
+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 From<strum::ParseError> for MinifiError {
- fn from(err: strum::ParseError) -> Self {
- MinifiError::Parse(ParseError::Strum(err))
+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 From<ParseBoolError> for MinifiError {
- fn from(err: ParseBoolError) -> Self {
- MinifiError::Parse(ParseError::Bool(err))
- }
+impl Error for RouteError {}
+
+#[derive(Debug)]
+pub enum ProcessError {
+ Route(RouteError),
+ Fatal(MinifiError),
}
-impl From<ParseIntError> for MinifiError {
- fn from(err: ParseIntError) -> Self {
- MinifiError::Parse(ParseError::Int(err))
+impl From<RouteError> for ProcessError {
+ fn from(err: RouteError) -> Self {
+ ProcessError::Route(err)
}
}
-impl From<humantime::DurationError> for MinifiError {
- fn from(err: humantime::DurationError) -> Self {
- MinifiError::Parse(ParseError::Duration(err))
+impl From<MinifiError> for ProcessError {
+ fn from(err: MinifiError) -> Self {
+ ProcessError::Fatal(err)
}
}
-impl From<byte_unit::ParseError> for MinifiError {
- fn from(err: byte_unit::ParseError) -> Self {
- MinifiError::Parse(ParseError::Size(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 From<NulError> for MinifiError {
- fn from(err: NulError) -> Self {
- MinifiError::Parse(ParseError::Nul(err))
+impl Error for ProcessError {}
+
+pub trait RouteErrorExt<T> {
+ fn route_err(self, rel: &Relationship, level: LogLevel) -> Result<T,
ProcessError>;
+
+ fn route_to(self, relationship: &'static str, 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(self, relationship_name: &'static str, level: LogLevel) ->
Result<T, ProcessError> {
+ self.map_err(|e| {
+ ProcessError::Route(RouteError {
+ relationship: relationship_name,
+ 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>),
+ UnscheduledProcessor,
+ ValidationError(Cow<'static, str>),
+ CustomError(Cow<'static, str>),
+ MissingFlowFileError,
+ IoError(std::io::Error),
+
+ Other(Box<dyn Error + Send + Sync + 'static>),
}
-impl From<ParseFloatError> for MinifiError {
- fn from(err: ParseFloatError) -> Self {
- MinifiError::Parse(ParseError::Float(err))
+impl From<std::io::Error> for MinifiError {
+ fn from(error: std::io::Error) -> Self {
+ MinifiError::IoError(error)
}
}
+macro_rules! minifi_error_from_validation {
+ ($($t:ty),* $(,)?) => {
+ $(
+ impl From<$t> for MinifiError {
+ fn from(err: $t) -> Self {
+ MinifiError::ValidationError(err.to_string().into())
+ }
+ }
+ )*
+ };
+}
+
+minifi_error_from_validation!(
+ strum::ParseError,
+ ParseBoolError,
+ ParseIntError,
+ humantime::DurationError,
+ byte_unit::ParseError,
+ NulError,
+ ParseFloatError,
+);
+
impl From<std::convert::Infallible> for MinifiError {
fn from(_: std::convert::Infallible) -> Self {
unreachable!("Infallible errors can never happen")
}
}
+impl From<anyhow::Error> for MinifiError {
+ fn from(err: anyhow::Error) -> Self {
+ Self::other(err)
+ }
+}
+
impl MinifiError {
pub(crate) fn to_status(&self) -> minifi_status {
match self {
@@ -116,40 +206,24 @@ impl MinifiError {
MinifiError::ValidationError(_) => {
minifi_native_sys::minifi_status_MINIFI_STATUS_VALIDATION_FAILED
}
- MinifiError::Parse(_) => {
-
minifi_native_sys::minifi_status_MINIFI_STATUS_VALIDATION_FAILED
- }
MinifiError::StatusError((_, ecode)) => u32::from(*ecode),
_ => minifi_native_sys::minifi_status_MINIFI_STATUS_UNKNOWN_ERROR,
}
}
- 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 custom<S: Into<Cow<'static, str>>>(msg: S) -> Self {
+ MinifiError::CustomError(msg.into())
}
- pub fn controller_service_err<S: Into<Cow<'static, str>>>(msg: S) -> Self {
- MinifiError::ControllerServiceError(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())
}
}
@@ -177,9 +251,76 @@ impl fmt::Display for MinifiError {
}
_ => write!(f, "{} (Unknown Status Code: {})", context, code),
},
+ MinifiError::Other(err) => write!(f, "{}", err),
+ MinifiError::ValidationError(msg) => write!(f, "{}", msg),
_ => 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, "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, "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..23587778e 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,41 @@
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;
pub struct GeneratedFlowFile<'a> {
target_relationship_name: &'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,
new_content,
- attributes_to_add,
+ attributes_to_add: Vec::new(),
}
}
- pub fn target_relationship_name(&self) -> &'static str {
+ 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 +59,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 +84,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 +101,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 +110,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 +125,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 +134,32 @@ 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())
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use crate::Relationship;
+ use minifi_native::GeneratedFlowFile;
+
+ const TEST_RELATIONSHIP: Relationship = Relationship {
+ name: "test",
+ description: "test desc",
+ };
+ #[test]
+ fn test_with_attributes() {
+ let mut gen_ff = GeneratedFlowFile::new(&TEST_RELATIONSHIP, None);
+ assert!(gen_ff.attributes_to_add.is_empty());
+
+ gen_ff = gen_ff.with_attribute("foo", "bar");
+ assert_eq!(1, gen_ff.attributes_to_add.len());
+
+ gen_ff = gen_ff.with_attributes([("A", "apple"), ("B", "banana")]);
+ assert_eq!(3, gen_ff.attributes_to_add.len());
+ let (key_1, value_1) = gen_ff.attributes_to_add.get(1).unwrap();
+ assert_eq!(key_1, "A");
+ assert_eq!(value_1, "apple");
+ }
+}
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..f300f99b8 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;
+#[derive(Debug)]
pub struct TransformStreamResult {
target_relationship_name: &'static str,
- attributes_to_add: HashMap<String, String>,
+ 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,
+ attributes_to_add: Vec::new(),
write_status: IoState::Ok,
}
}
pub fn route_without_changes(target_relationship: &Relationship) -> Self {
+ Self::route_without_changes_by_name(target_relationship.name)
+ }
+
+ pub fn route_without_changes_by_name(relationship: &'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 {
+ 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()
+ .rfind(|(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,32 @@ 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())
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use crate::Relationship;
+ use minifi_native::TransformStreamResult;
+
+ const TEST_RELATIONSHIP: Relationship = Relationship {
+ name: "test",
+ description: "test desc",
+ };
+ #[test]
+ fn test_with_attributes() {
+ let mut gen_ff = TransformStreamResult::new(&TEST_RELATIONSHIP);
+ assert!(gen_ff.attributes_to_add.is_empty());
+
+ gen_ff = gen_ff.with_attribute("foo", "bar");
+ assert_eq!(1, gen_ff.attributes_to_add.len());
+
+ gen_ff = gen_ff.with_attributes([("A", "apple"), ("B", "banana")]);
+ assert_eq!(3, gen_ff.attributes_to_add.len());
+ let (key_1, value_1) = gen_ff.attributes_to_add.get(1).unwrap();
+ assert_eq!(key_1, "A");
+ assert_eq!(value_1, "apple");
+ }
+}
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..2f470125e 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,
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(target_relationship.name)
+ }
+
+ pub fn route_without_changes_by_name(relationship: &'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<Content<'a>>) -> Self {
Self {
target_relationship_name: target_relationship.name,
- new_content: new_content.map(Content::Buffer),
- attributes_to_add,
+ new_content,
+ attributes_to_add: Vec::new(),
}
}
+ #[must_use]
+ pub fn with_content(mut self, content: Content<'a>) -> Self {
+ self.new_content = Some(content);
+ self
+ }
+
pub fn new_content(&'_ self) -> Option<&'_ Content<'_>> {
self.new_content.as_ref()
}
- pub fn target_relationship(&self) -> &'static str {
+ 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()
+ .rfind(|(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,141 @@ 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);
+ }
+
+ const TEST_RELATIONSHIP: Relationship = Relationship {
+ name: "test",
+ description: "test desc",
};
+ #[test]
+ fn test_with_attributes() {
+ let mut gen_ff =
TransformedFlowFile::route_without_changes(&TEST_RELATIONSHIP);
+ assert!(gen_ff.attributes_to_add.is_empty());
+
+ gen_ff = gen_ff.with_attribute("foo", "bar");
+ assert_eq!(1, gen_ff.attributes_to_add.len());
+
+ gen_ff = gen_ff.with_attributes([("A", "apple"), ("B", "banana")]);
+ assert_eq!(3, gen_ff.attributes_to_add.len());
+ let (key_1, value_1) = gen_ff.attributes_to_add.get(1).unwrap();
+ assert_eq!(key_1, "A");
+ assert_eq!(value_1, "apple");
+
+ assert!(gen_ff.new_content.is_none());
+ gen_ff = gen_ff.with_content(Content::Buffer("hello".into()));
+ assert!(matches!(gen_ff.new_content.unwrap(), Content::Buffer(_)));
+ }
}
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..fc924f07f
--- /dev/null
+++
b/minifi_rust/minifi_native/src/api/processor_wrappers/utils/with_attributes.rs
@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/// Adds implementation with_attribute and with_attribute(s)
+/// used by TransformedFlowFile<'a>, GeneratedFlowFile<'a>,
TransformStreamResult
+/// extracted to macro to avoid code duplication
+#[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 1e85c39bd..9c70fa016 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..9bed92bd6 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;
@@ -546,10 +549,10 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
}
if let Some(result) = ctx.result.take() {
- result
- } else {
- Err(MinifiError::UnknownError)
+ return result;
}
+
+ 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..4114b524f 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)
}
}
}
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;
