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 eb14c120f23ef186b92a0ff7ce30089355e5fde1 Author: Martin Zink <[email protected]> AuthorDate: Tue Aug 18 09:43:20 2026 +0200 review changes --- .../minifi_rs_playground/src/processors/asciify_german/tests.rs | 5 ++++- minifi_rust/minifi_native/src/api/errors.rs | 2 +- .../src/api/processor_wrappers/flow_file_stream_transform.rs | 3 ++- .../src/api/processor_wrappers/flow_file_transform.rs | 2 +- minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs | 8 ++++---- 5 files changed, 12 insertions(+), 8 deletions(-) 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 1a2a77d60..6ee76de72 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 @@ -84,5 +84,8 @@ fn truncated_umlaut_at_eof_routes_to_failure() { let mut output_vec: Vec<u8> = Vec::new(); let result = asciify_german.transform(&context, &mut input_stream, &mut output_vec, &logger); - assert!(result.is_err()); + match result { + Err(ProcessError::Route(route)) => assert_eq!(route.relationship.as_ref(), FAILURE.name), + other => panic!("expected a route error to failure, got {other:?}"), + } } diff --git a/minifi_rust/minifi_native/src/api/errors.rs b/minifi_rust/minifi_native/src/api/errors.rs index 9e5749fcc..b2b4230d6 100644 --- a/minifi_rust/minifi_native/src/api/errors.rs +++ b/minifi_rust/minifi_native/src/api/errors.rs @@ -259,7 +259,7 @@ impl fmt::Display for MinifiError { } _ => write!(f, "{} (Unknown Status Code: {})", context, code), }, - MinifiError::Other(err) => write!(f, "Custom error: {}", err), + MinifiError::Other(err) => write!(f, "{}", err), MinifiError::ValidationError(msg) => write!(f, "{}", msg), _ => write!(f, "{:?}", self), } 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 5b6a2643a..52a6fe0db 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 @@ -26,6 +26,7 @@ use crate::{ }; use std::borrow::Cow; +#[derive(Debug)] pub struct TransformStreamResult { target_relationship_name: Cow<'static, str>, attributes_to_add: Vec<FlowFileAttribute>, @@ -60,7 +61,7 @@ impl TransformStreamResult { pub fn get_attribute(&self, name: &str) -> Option<&str> { self.attributes_to_add .iter() - .find(|(k, _)| k == name) + .rfind(|(k, _)| k == name) .map(|(_, v)| v.as_ref()) } 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 462b70973..30e68599f 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 @@ -81,7 +81,7 @@ impl<'a> TransformedFlowFile<'a> { pub fn attribute(&self, name: &str) -> Option<&str> { self.attributes_to_add .iter() - .find(|(k, _)| k == name) + .rfind(|(k, _)| k == name) .map(|(_, v)| v.as_ref()) } 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 a2207ca85..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 @@ -541,10 +541,6 @@ 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(), @@ -552,6 +548,10 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { ))); } + if let Some(result) = ctx.result.take() { + return result; + } + Err(MinifiError::UnknownError) } }
