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
The following commit(s) were added to refs/heads/minifi_rust_impr_2 by this
push:
new 2b1560f29 fixes
2b1560f29 is described below
commit 2b1560f29156d417bb93253a859e3cd6ba36ef7a
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 | 44 ++++++++++------------
minifi_rust/minifi_native/src/mock.rs | 1 +
.../src/mock/mock_resolve_process_err.rs | 0
minifi_rust/minifi_native_sys/build.rs | 3 +-
5 files changed, 25 insertions(+), 27 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..054544307 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
@@ -21,8 +21,8 @@ use crate::processors::put_file::definitions::{FAILURE,
SUCCESS};
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, ProcessError, RouteErrorExt, Schedule, TransformedFlowFile,
trace, warn,
+ trace, warn, FlowFileTransform, GetAttribute, GetControllerService, GetId,
GetProperty,
+ InputStream, Logger, MinifiError, ProcessError, RouteErrorExt, Schedule,
TransformedFlowFile,
};
use std::path::{Path, PathBuf};
use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
@@ -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/src/mock.rs
b/minifi_rust/minifi_native/src/mock.rs
index 49e7c7c8d..a3260bfb9 100644
--- a/minifi_rust/minifi_native/src/mock.rs
+++ b/minifi_rust/minifi_native/src/mock.rs
@@ -20,6 +20,7 @@ mod mock_flow_file;
mod mock_logger;
mod mock_process_context;
mod mock_process_session;
+mod mock_resolve_process_err;
pub use mock_controller_service_context::MockControllerServiceContext;
pub use mock_flow_file::MockFlowFile;
diff --git a/minifi_rust/minifi_native/src/mock/mock_resolve_process_err.rs
b/minifi_rust/minifi_native/src/mock/mock_resolve_process_err.rs
new file mode 100644
index 000000000..e69de29bb
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");
