szaszm commented on code in PR #2186: URL: https://github.com/apache/nifi-minifi-cpp/pull/2186#discussion_r3634399769
########## minifi_rust/minifi_native/src/api/process_context.rs: ########## @@ -0,0 +1,151 @@ +// 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. + +use crate::StandardPropertyValidator::*; +use crate::api::RawControllerService; +use crate::api::component_definition_traits::ComponentIdentifier; +use crate::api::flow_file::FlowFile; +use crate::api::property::GetControllerService; +use crate::{ + ControllerServiceApi, ControllerServiceDefinition, EnableControllerService, GetProperty, + MinifiError, Property, +}; +use std::str::FromStr; +use std::time::Duration; + +pub trait ProcessContext { + type FlowFile: FlowFile; + + fn get_property( + &self, + property: &Property, + flow_file: Option<&Self::FlowFile>, + ) -> Result<Option<String>, MinifiError>; + + fn get_bool_property( + &self, + property: &Property, + flow_file: Option<&Self::FlowFile>, + ) -> Result<Option<bool>, MinifiError> { + if property.validator != BoolValidator { + return Err(MinifiError::validation_err(format!( + "to use get_bool_property {:?} must have BoolValidator", + property + ))); + } + + if let Some(property_val) = self.get_property(property, flow_file)? { + Ok(Some(bool::from_str(&property_val)?)) + } else { + Ok(None) + } + } + + fn get_duration_property( + &self, + property: &Property, + flow_file: Option<&Self::FlowFile>, + ) -> Result<Option<Duration>, MinifiError> { + if property.validator != TimePeriodValidator { + return Err(MinifiError::validation_err(format!( + "to use get_duration_property {:?} must have TimePeriodValidator", + property + ))); + } + + if let Some(property_val) = self.get_property(property, flow_file)? { + Ok(Some(humantime::parse_duration(property_val.as_str())?)) + } else { + Ok(None) + } + } + + fn get_size_property( + &self, + property: &Property, + flow_file: Option<&Self::FlowFile>, + ) -> Result<Option<u64>, MinifiError> { + if property.validator != DataSizeValidator { + return Err(MinifiError::validation_err(format!( + "to use get_size_property {:?} must have DataSizeValidator", + property + ))); + } + if let Some(property_val) = self.get_property(property, flow_file)? { + Ok(Some(byte_unit::Byte::from_str(&property_val)?.as_u64())) Review Comment: Are the units recognized by this byte_unit parser compatible with NiFi and MiNiFi C++ unit conventions? Namely that MB means MiB, the unit part is case insensitive, and the M suffix means 1000000 (power of 10 rather than 2, unlike in Unix tools) ########## minifi_rust/extensions/minifi_rs_playground/src/processors/get_file.rs: ########## @@ -0,0 +1,327 @@ +// 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. + +use crate::processors::get_file::output_attributes::{ + ABSOLUTE_PATH_OUTPUT_ATTRIBUTE, FILENAME_OUTPUT_ATTRIBUTE, +}; +use crate::processors::get_file::properties::{ + BATCH_SIZE, DIRECTORY, IGNORE_HIDDEN_FILES, KEEP_SOURCE_FILE, MAX_AGE, MAX_SIZE, MIN_AGE, + MIN_SIZE, RECURSE, +}; +use minifi_native::macros::ComponentIdentifier; +use minifi_native::{ + GetProperty, IoState, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, + Schedule, Trigger, debug, info, trace, warn, +}; +use std::collections::VecDeque; +use std::error; +use std::fs::File; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::{Duration, Instant, SystemTime}; +use walkdir::{DirEntry, WalkDir}; + +mod properties; +mod relationships; + +#[derive(Debug)] +struct GetFileMetrics { + accepted_files: u32, + input_bytes: u64, +} + +#[derive(Debug)] +struct DirectoryListing { + paths: VecDeque<PathBuf>, + last_polling_time: Option<Instant>, +} + +impl DirectoryListing { + fn new() -> Self { + Self { + paths: VecDeque::new(), + last_polling_time: None, + } + } +} + +#[derive(Debug, ComponentIdentifier)] +pub(crate) struct GetFileRs { + recursive: bool, + keep_source_file: bool, + input_directory: PathBuf, + poll_interval: Option<Duration>, + directory_listing: Mutex<DirectoryListing>, + batch_size: u64, + min_size: Option<u64>, + max_size: Option<u64>, + min_age: Option<Duration>, + max_age: Option<Duration>, + ignore_hidden_files: bool, + metrics: Mutex<GetFileMetrics>, +} + +impl GetFileRs { + fn is_listing_empty(&self) -> bool { + let directory_listing = self.directory_listing.lock().unwrap(); + directory_listing.paths.is_empty() + } + + fn poll_listing(&self, batch_size: u64) -> VecDeque<PathBuf> { + let mut directory_listings = self.directory_listing.lock().unwrap(); + + let mut res = VecDeque::new(); + for _ in 0..batch_size { + if let Some(path) = directory_listings.paths.pop_back() { + res.push_back(path); + } else { + break; + } + } + Review Comment: If batch_size is large enough (which I think in most cases it is), isn't this just a move? Also this approach will reverse the order of elements, hopefully that's not a problem. ########## minifi_rust/README.md: ########## @@ -0,0 +1,96 @@ +# MiNiFi Native Rust (tech preview) + +> ⚠️ **Tech Preview** +> +> Only the **Rust bindings** are a technology preview. The Rust source API +> (both the `minifi-native` safe API and the `minifi-native-sys` FFI layer) +> is still evolving, and we are **not yet committed to source-level backward +> compatibility** for the Rust bindings — expect breaking changes in the +> Rust API between MiNiFi C++ releases until the bindings are declared +> stable. +> +> Under the hood, the bindings target the **stable MiNiFi C API**, which +> *does* provide backward compatibility. This means the compiled artifact Review Comment: ```suggestion > *does* provide ABI backward compatibility. This means the compiled artifact ``` ########## minifi_rust/minifi_native/src/api/processor_wrappers/flow_file_transform.rs: ########## @@ -0,0 +1,202 @@ +// 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. + +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; +use crate::api::processor_wrappers::utils::flow_file_content::Content; +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, +}; +use std::collections::HashMap; + +#[derive(Debug)] +pub struct TransformedFlowFile<'a> { + target_relationship_name: &'static str, + new_content: Option<Content<'a>>, // If None the content doesn't change Review Comment: ```suggestion new_content: Option<Content<'a>>, // If None, the content doesn't change ``` ########## minifi_rust/minifi_native/src/c_ffi/c_ffi_primitives.rs: ########## @@ -0,0 +1,103 @@ +// 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. + +use crate::ProcessorInputRequirement; +use minifi_native_sys::{ + minifi_input_requirement, minifi_input_requirement_MINIFI_INPUT_ALLOWED, + minifi_input_requirement_MINIFI_INPUT_FORBIDDEN, + minifi_input_requirement_MINIFI_INPUT_REQUIRED, minifi_string_view, +}; +use std::os::raw::c_char; + +#[derive(Debug)] +pub enum FfiConversionError { + NullPointer, + InvalidUtf8, +} + +#[repr(transparent)] // This allows us to pass Vec<StringView> as a pointer to minifi_string_view array +#[derive(Debug)] +pub(crate) struct StringView<'a> { + inner: minifi_string_view, + _marker: std::marker::PhantomData<&'a ()>, +} + +impl<'a> StringView<'a> { + pub(crate) fn new(str: &'a str) -> Self { + Self { + inner: minifi_string_view { + data: str.as_ptr() as *const c_char, + length: str.len(), + }, + _marker: std::marker::PhantomData, + } + } + + pub unsafe fn as_raw(&self) -> minifi_string_view { + self.inner + } +} + +pub trait StaticStrAsMinifiCStr { + fn as_minifi_c_type(&self) -> minifi_string_view; +} + +impl StaticStrAsMinifiCStr for &'static str { + fn as_minifi_c_type(&self) -> minifi_string_view { + minifi_string_view { + data: self.as_ptr() as *const c_char, + length: self.len(), + } + } +} + +pub trait ConvertMinifiStringView { + unsafe fn as_string(&self) -> Result<String, FfiConversionError>; + unsafe fn as_str(&self) -> Result<&str, FfiConversionError>; Review Comment: Wouldn't it be better to implement as_string in terms of as_str and to_owned? Only one unsafe function instead of two that way. ########## minifi_rust/README.md: ########## @@ -0,0 +1,96 @@ +# MiNiFi Native Rust (tech preview) + +> ⚠️ **Tech Preview** +> +> Only the **Rust bindings** are a technology preview. The Rust source API +> (both the `minifi-native` safe API and the `minifi-native-sys` FFI layer) +> is still evolving, and we are **not yet committed to source-level backward +> compatibility** for the Rust bindings — expect breaking changes in the +> Rust API between MiNiFi C++ releases until the bindings are declared +> stable. +> +> Under the hood, the bindings target the **stable MiNiFi C API**, which +> *does* provide backward compatibility. This means the compiled artifact +> is unaffected by Rust-side churn: an extension built against an older +> version of these bindings will keep loading into newer MiNiFi C++ +> releases. You only need to rebuild against the new bindings if you want +> to pick up new Rust API features — upgrading the agent alone does not +> force a rebuild. + +This repository provides a safe, idiomatic, and high-performance Rust framework for building native extensions (processors) for [Apache NiFi MiNiFi C++](https://github.com/apache/nifi-minifi-cpp). + +It is designed to offer a robust developer experience, allowing you to write powerful and reliable data processing components in safe Rust. + +The framework completely encapsulates the unsafe C FFI (Foreign Function Interface) boundary, providing a pure Rust API that is fully mockable for unit testing. + +## Project Philosophy + - **Safety First**: Leverage Rust's compile-time guarantees to prevent common bugs like null pointers, buffer overflows, and data races. + - **Zero-Cost Abstraction**: The safe API wrapper is designed to compile down with zero runtime overhead compared to writing raw C code. + - **Ergonomics**: Provide a clean, idiomatic Rust API that is a pleasure to use. Developers should not need to think about unsafe code or C++ interoperability. + - **Testability**: Every component of a processor's logic should be unit-testable in a pure Rust environment, without needing a C++ host. + - **Cross Platform**: The library should work on all platforms that are supported by [Apache NiFi MiNiFi C++](https://github.com/apache/nifi-minifi-cpp). + - macOS (aarch64) + - Linux (x86_64, aarch64) + - Windows (x86_64) + +The project is structured as a Cargo workspace with a clear, layered architecture: +### [minifi-native-sys](minifi_native_sys) +Contains the raw, unsafe FFI bindings to the minifi-api.h C API. +### [minifi-native](minifi_native) +Provides the public, safe, and idiomatic Rust API. This is the crate that developers will use to build their processors. +#### API Traits +Pure Rust traits (Processor, ProcessSession, Logger, etc.) that define the abstract behavior of the MiNiFi environment. +#### Higher level API +Pure rust traits that simplify the requirements for a working processor. Pick the one that matches your processor's shape — the wrapper takes care of getting/creating the flowfile, wiring up streams, applying attributes, and transferring to the right relationship, so your code only needs to describe the transformation itself. + - [FlowFileTransform](minifi_native/src/api/processor_wrappers/flow_file_transform.rs) + - Consumes a single incoming flowfile, optionally rewrites its content (returned as an in-memory buffer) and/or adds attributes, then routes it to a relationship. Best for buffered, one-in / one-out transforms. Review Comment: Couldn't you also wrap the input stream to achieve streaming transform with FlowFileTransform? Because it gets an InputStream and can return a Stream containing a Box<dyn Read>. If that doesn't work, where does the full reading into memory happen? ########## minifi_rust/README.md: ########## Review Comment: maybe it's just me, but I'd leave an empty line before headings for better visual separation in the source text ########## minifi_rust/minifi_native/src/lib.rs: ########## @@ -0,0 +1,135 @@ +// 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. + +mod api; +pub mod c_ffi; +pub mod mock; + +pub use api::errors::MinifiError; + +pub use api::component_definition_traits::{ + ComponentIdentifier, ControllerServiceDefinition, ProcessorDefinition, +}; +pub use api::controller_service::{ControllerService, EnableControllerService}; +pub use api::processor_wrappers::complex_processor::{ComplexProcessorType, MutTrigger, Trigger}; +pub use api::processor_wrappers::flow_file_source::{ + FlowFileSource, FlowFileSourceProcessorType, GeneratedFlowFile, +}; +pub use api::processor_wrappers::flow_file_stream_transform::{ + FlowFileStreamTransform, FlowFileStreamTransformProcessorType, MutFlowFileStreamTransform, + TransformStreamResult, +}; +pub use api::processor_wrappers::flow_file_transform::{ + FlowFileTransform, FlowFileTransformProcessorType, TransformedFlowFile, +}; + +pub use api::processor_wrappers::utils::flow_file_content::Content; + +pub use api::processor::{Processor, Schedule}; + +pub use api::raw_processor::{MultiThreaded, SingleThreaded}; + +pub use api::logger::{LogLevel, Logger}; + +pub use api::property::{GetControllerService, GetProperty, Property}; + +pub use api::provided_interface::{ControllerServiceApi, ProvidedInterface}; + +pub use api::process_session::IoState; + +pub use api::attribute::{GetAttribute, OutputAttribute}; + +pub use api::{ + FlowFile, GetId, InputStream, OnTriggerResult, OutputStream, ProcessContext, ProcessSession, + ProcessorInputRequirement, Relationship, StandardPropertyValidator, +}; + +pub use minifi_native_macros as macros; +pub use minifi_native_sys as sys; +pub use mock::{ + MockControllerServiceContext, MockFlowFile, MockLogger, MockProcessContext, MockProcessSession, + StdLogger, +}; + +#[unsafe(no_mangle)] +#[allow(non_upper_case_globals)] +#[cfg_attr(target_os = "linux", unsafe(link_section = ".rodata"))] +#[cfg_attr(target_os = "macos", unsafe(link_section = "__DATA,__const"))] +#[cfg_attr(target_os = "windows", unsafe(link_section = ".rdata"))] +pub static minifi_api_version: u32 = minifi_native_sys::MINIFI_API_VERSION; Review Comment: no extern "c" or repr(C) necessary here? would const work here, or could that not use the macro symbol from the bindings? ########## minifi_rust/README.md: ########## @@ -0,0 +1,96 @@ +# MiNiFi Native Rust (tech preview) + +> ⚠️ **Tech Preview** +> +> Only the **Rust bindings** are a technology preview. The Rust source API +> (both the `minifi-native` safe API and the `minifi-native-sys` FFI layer) +> is still evolving, and we are **not yet committed to source-level backward +> compatibility** for the Rust bindings — expect breaking changes in the +> Rust API between MiNiFi C++ releases until the bindings are declared +> stable. +> +> Under the hood, the bindings target the **stable MiNiFi C API**, which +> *does* provide backward compatibility. This means the compiled artifact +> is unaffected by Rust-side churn: an extension built against an older +> version of these bindings will keep loading into newer MiNiFi C++ +> releases. You only need to rebuild against the new bindings if you want +> to pick up new Rust API features — upgrading the agent alone does not +> force a rebuild. + +This repository provides a safe, idiomatic, and high-performance Rust framework for building native extensions (processors) for [Apache NiFi MiNiFi C++](https://github.com/apache/nifi-minifi-cpp). Review Comment: ```suggestion This project provides a safe, idiomatic, and high-performance Rust framework for building native extensions (processors) for [Apache NiFi MiNiFi C++](https://github.com/apache/nifi-minifi-cpp). ``` -- 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]
