Copilot commented on code in PR #2246:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2246#discussion_r3777197019
##########
minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs:
##########
@@ -538,18 +541,18 @@ impl<'a> ProcessSession for CffiProcessSession<'a> {
&mut ctx as *mut _ as *mut c_void,
);
+ if let Some(result) = ctx.result.take() {
+ return result;
Review Comment:
A successful callback result is returned before the C session status is
checked, so an exception reported by `minifi_process_session_read` after
invoking the callback is silently converted to success. Preserve callback
errors, but validate a non-success status before returning an `Ok` callback
value.
##########
minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs:
##########
@@ -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)
Review Comment:
Repeated `with_attribute` calls for the same key are applied in order, so
the last value becomes the FlowFile attribute, but this accessor returns the
first value. Search from the end so this API reports the value that will
actually be applied.
##########
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());
Review Comment:
This assertion now accepts every error variant, so it no longer verifies the
behavior named by the test: routing specifically to `failure`. A fatal
processing error or a route to the wrong relationship would pass unnoticed.
##########
minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_stream_transform.rs:
##########
@@ -18,60 +18,67 @@
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)
Review Comment:
Repeated `with_attribute` calls for the same key are applied in order, so
the last value becomes the FlowFile attribute, but this getter returns the
first value. Search from the end to keep the observable result consistent with
`set_attribute`.
##########
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());
Review Comment:
This weakens the previous variant-specific check to accept any scheduling
failure, so regressions such as a missing-property error would satisfy the
test. The implementation now deliberately returns `CustomError`; retain that
contract in the assertion.
##########
minifi_rust/minifi_native/src/api/errors.rs:
##########
@@ -177,9 +293,75 @@ impl fmt::Display for MinifiError {
}
_ => write!(f, "{} (Unknown Status Code: {})", context, code),
},
+ MinifiError::Other(err) => write!(f, "Custom error: {}", err),
Review Comment:
`Other` wraps an arbitrary source error, but its display text labels every
such failure as a custom error even though `CustomError` is a separate variant.
This produces misleading operational diagnostics; display the wrapped error
transparently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
