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 ee71fc6d211f6d0d08d3c6975dc5d49e4503353f Author: Martin Zink <[email protected]> AuthorDate: Thu Aug 13 13:47:14 2026 +0200 fixes --- .../features/error-handling.feature | 4 +-- .../src/processors/put_file.rs | 40 ++++++++++------------ minifi_rust/minifi_native_sys/build.rs | 3 +- 3 files changed, 22 insertions(+), 25 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/put_file.rs b/minifi_rust/extensions/minifi_rs_playground/src/processors/put_file.rs index 47292e04a..e20337f8f 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 @@ -50,21 +50,23 @@ pub(crate) struct PutFileRs { } impl PutFileRs { - pub(crate) fn directory_is_full(&self, p0: &Path) -> bool { + pub(crate) fn check_for_full_dir(&self, p0: &Path) -> Result<(), MinifiError> { if let Some(max_file_count) = self.maximum_file_count && let Some(parent) = p0.parent() { - parent.exists() + let full_dir = parent.exists() && WalkDir::new(parent) .max_depth(1) .into_iter() .filter_map(Result::ok) .filter(|e| e.file_type().is_file()) .count() - >= max_file_count as usize - } else { - false + >= max_file_count as usize; + if full_dir { + return Err(MinifiError::custom("Directory is full")); + } } + Ok(()) } fn get_destination_path<Ctx>(context: &Ctx) -> Result<PathBuf, MinifiError> @@ -170,11 +172,8 @@ impl FlowFileTransform for PutFileRs { trace!(logger, "on_trigger: {:?}", self); let destination_path = Self::get_destination_path(context).route_err_to_failure()?; - - if self.directory_is_full(&destination_path) { - warn!(logger, "Directory is full"); - return Ok(TransformedFlowFile::route_without_changes(&FAILURE)); - } + self.check_for_full_dir(&destination_path) + .route_err_to_failure()?; if destination_path.exists() { match self.conflict_resolution_strategy { @@ -190,10 +189,9 @@ impl FlowFileTransform for PutFileRs { } } - match self.put_file(input_stream, logger, &destination_path) { - Ok(_) => Ok(TransformedFlowFile::route_without_changes(&SUCCESS)), - Err(_e) => Ok(TransformedFlowFile::route_without_changes(&FAILURE)), - } + self.put_file(input_stream, logger, &destination_path) + .route_err_to_failure()?; + Ok(TransformedFlowFile::route_without_changes(&SUCCESS)) } } @@ -257,11 +255,9 @@ mod tests { context .attributes .insert("filename".to_string(), "test.txt".to_string()); - let result = put_file - .transform(&context, &mut input_stream, &MockLogger::new()) - .expect("Should succeed"); + let result = put_file.transform(&context, &mut input_stream, &MockLogger::new()); - assert_eq!(result.target_relationship(), FAILURE.name); + assert!(result.is_err_and(|e| { matches!(e, ProcessError::Route(_)) })); let expected_path = temp_dir.path().join("subdir/test.txt"); assert!(!expected_path.exists()); @@ -285,20 +281,20 @@ mod tests { let destination = temp_dir.path().join("test.txt"); // No files yet → not full - assert!(!put_file.directory_is_full(&destination)); + assert!(put_file.check_for_full_dir(&destination).is_ok()); // Create a subdirectory; it must not be counted as a file std::fs::create_dir(temp_dir.path().join("subdir")).unwrap(); - assert!(!put_file.directory_is_full(&destination)); + assert!(put_file.check_for_full_dir(&destination).is_ok()); // Add two files → full std::fs::write(temp_dir.path().join("a.txt"), b"a").unwrap(); std::fs::write(temp_dir.path().join("b.txt"), b"b").unwrap(); - assert!(put_file.directory_is_full(&destination)); + assert!(put_file.check_for_full_dir(&destination).is_err()); // Remove one file → not full again std::fs::remove_file(temp_dir.path().join("a.txt")).unwrap(); - assert!(!put_file.directory_is_full(&destination)); + assert!(put_file.check_for_full_dir(&destination).is_ok()); } #[cfg(unix)] 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");
