diff --git a/rust/crates/eckit-sys/cpp/DataHandleWrapper.cc b/rust/crates/eckit-sys/cpp/DataHandleWrapper.cc index 2b9fbeade..23c021575 100644 --- a/rust/crates/eckit-sys/cpp/DataHandleWrapper.cc +++ b/rust/crates/eckit-sys/cpp/DataHandleWrapper.cc @@ -153,6 +153,55 @@ class RustReaderHandle : public eckit::DataHandle { rust::Box reader_; }; +// `eckit::DataHandle` subclass that forwards `write()` to a Rust `Write` +// sink held in a `rust::Box`. `close()` flushes the sink. The +// handle is write-only: the read-side virtuals keep the base class +// behaviour of throwing. +class RustWriterHandle : public eckit::DataHandle { +public: + + explicit RustWriterHandle(rust::Box writer) : writer_(std::move(writer)) {} + + void print(std::ostream& s) const override { s << "RustWriterHandle[]"; } + + void openForWrite(const eckit::Length&) override {} + + long write(const void* buffer, long length) override { + if (length <= 0) { + return 0; + } + const auto* bytes = static_cast(buffer); + rust::Slice slice{bytes, static_cast(length)}; + int64_t n = invoke_writer_write(*writer_, slice); + if (n < 0) { + throw eckit::WriteError("RustWriterHandle: error writing to Rust sink"); + } + return static_cast(n); + } + + bool canSeek() const override { return false; } + + void flush() override { + if (invoke_writer_flush(*writer_) < 0) { + throw eckit::WriteError("RustWriterHandle: flush failed"); + } + } + + void close() override { + if (invoke_writer_flush(*writer_) < 0) { + throw eckit::WriteError("RustWriterHandle: flush failed on close"); + } + } + + eckit::Length estimate() override { return eckit::Length(0); } + + eckit::Length size() override { return eckit::Length(0); } + +private: + + rust::Box writer_; +}; + } // namespace //---------------------------------------------------------------------------------------------------------------------- @@ -161,6 +210,13 @@ std::unique_ptr DataHandleWrapper::from_reader(rust::Box(new RustReaderHandle(std::move(reader))); } +std::unique_ptr DataHandleWrapper::from_writer(rust::Box writer) { + auto handle = std::make_unique(std::move(writer)); + auto wrapper = std::make_unique(handle.get()); + handle.release(); + return wrapper; +} + //---------------------------------------------------------------------------------------------------------------------- } // namespace eckit_bridge diff --git a/rust/crates/eckit-sys/cpp/DataHandleWrapper.h b/rust/crates/eckit-sys/cpp/DataHandleWrapper.h index 888547f67..8244c32c1 100644 --- a/rust/crates/eckit-sys/cpp/DataHandleWrapper.h +++ b/rust/crates/eckit-sys/cpp/DataHandleWrapper.h @@ -15,8 +15,9 @@ namespace eckit_bridge { //---------------------------------------------------------------------------------------------------------------------- -// Forward declaration — defined on the Rust side, cxx generates the type. +// Forward declarations — defined on the Rust side, cxx generates the types. struct ReaderBox; +struct WriterBox; /// Wraps `eckit::DataHandle*` for Rust FFI. Takes ownership. class DataHandleWrapper { @@ -75,6 +76,16 @@ class DataHandleWrapper { /// The returned handle is owned by the Rust side; on drop, the contained /// `ReaderBox` is dropped, releasing the underlying `Read` source. static std::unique_ptr from_reader(rust::Box reader); + + /// Create a DataHandle that forwards `write()` calls to a Rust + /// `std::io::Write` sink wrapped in a `WriterBox`. Used to stream bytes + /// from any C++ API that produces into an `eckit::DataHandle&` (e.g. + /// the odc encoder) directly into a Rust sink, without staging through + /// a temp file or buffer. + /// + /// The returned handle is owned by the Rust side; on drop, the contained + /// `WriterBox` is dropped, releasing the underlying `Write` sink. + static std::unique_ptr from_writer(rust::Box writer); }; //---------------------------------------------------------------------------------------------------------------------- diff --git a/rust/crates/eckit-sys/src/lib.rs b/rust/crates/eckit-sys/src/lib.rs index 3336b4794..e2f758cf2 100644 --- a/rust/crates/eckit-sys/src/lib.rs +++ b/rust/crates/eckit-sys/src/lib.rs @@ -159,6 +159,11 @@ mod ffi { #[Self = "DataHandleWrapper"] fn from_reader(reader: Box) -> Result>; + /// Create a DataHandle that forwards `write()` calls to a Rust + /// `std::io::Write` sink wrapped in a [`WriterBox`]. + #[Self = "DataHandleWrapper"] + fn from_writer(writer: Box) -> Result>; + // ==================== Message + Reader ==================== type MessageWrapper; @@ -271,6 +276,26 @@ mod ffi { /// /// Returns the new absolute position, or `-1` on error. fn invoke_reader_seek(reader: &mut ReaderBox, offset: i64) -> i64; + + /// Opaque Rust box holding a `dyn Write + Send` sink. + /// + /// Constructed via [`make_writer_box`]; the C++ `RustWriterHandle` + /// holds it by `rust::Box` and forwards `write()` and + /// `close()` calls through [`invoke_writer_write`] / + /// [`invoke_writer_flush`]. + type WriterBox; + + /// Called by the C++ `RustWriterHandle::write` shim to forward the + /// next chunk to the wrapped Rust `Write` sink. + /// + /// Returns the chunk length on success, or `-1` on error. + fn invoke_writer_write(writer: &mut WriterBox, buf: &[u8]) -> i64; + + /// Called by the C++ `RustWriterHandle::close` shim to flush the + /// wrapped Rust sink. + /// + /// Returns `0` on success, or `-1` on error. + fn invoke_writer_flush(writer: &mut WriterBox) -> i64; } } @@ -330,6 +355,42 @@ fn invoke_reader_seek(reader: &mut ReaderBox, offset: i64) -> i64 { .map_or(-1, |n| i64::try_from(n).unwrap_or(i64::MAX)) } +// ==================== Write → DataHandle adapter ==================== + +/// Opaque wrapper holding a `Box`. +/// +/// The C++ `RustWriterHandle` (declared in `DataHandleWrapper.h` as `struct +/// WriterBox`) carries this by `rust::Box` and forwards each C++ +/// `write(const void*, long)` call via `invoke_writer_write`. +pub struct WriterBox(Box); + +/// Wrap a Rust `Write` sink for `ffi::from_writer`. +pub fn make_writer_box(writer: W) -> Box +where + W: std::io::Write + Send + 'static, +{ + Box::new(WriterBox(Box::new(writer))) +} + +/// Called from C++ `RustWriterHandle::write` to forward the next chunk to +/// the wrapped Rust sink. The whole chunk is consumed. Returns the chunk +/// length, or `-1` on error — including a panicking sink, since the C++ +/// caller treats any failed write as fatal to the transfer (partial +/// progress is never retried, the resulting exception aborts it). +fn invoke_writer_write(writer: &mut WriterBox, buf: &[u8]) -> i64 { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| writer.0.write_all(buf))) + .map_or(-1, |result| { + result.map_or(-1, |()| i64::try_from(buf.len()).unwrap_or(i64::MAX)) + }) +} + +/// Called from C++ `RustWriterHandle::close` to flush the wrapped sink. +/// Returns `0` on success, or `-1` on error, including a panicking sink. +fn invoke_writer_flush(writer: &mut WriterBox) -> i64 { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| writer.0.flush())) + .map_or(-1, |result| result.map_or(-1, |()| 0)) +} + /// Called from C++ `RustLogTarget::write()`; routes to Rust `log` crate. fn rust_log(level: ffi::LogLevel, target: &str, msg: &str) { match level { diff --git a/rust/crates/eckit/src/datahandle.rs b/rust/crates/eckit/src/datahandle.rs index a534df77d..16450e833 100644 --- a/rust/crates/eckit/src/datahandle.rs +++ b/rust/crates/eckit/src/datahandle.rs @@ -134,6 +134,21 @@ impl DataHandle { Ok(Self::from_raw(inner)) } + /// Wrap a Rust `std::io::Write` sink as a `DataHandle`. + /// + /// The C++ side calls back into the Rust sink on each `write()`; no + /// intermediate buffer or temp file is staged. Closing the handle + /// flushes the sink, and dropping it drops the sink. The resulting + /// handle is write-only; it cannot be read or seeked. + pub fn from_writer(writer: W) -> Result + where + W: std::io::Write + Send + 'static, + { + let inner = eckit_sys::DataHandleWrapper::from_writer(eckit_sys::make_writer_box(writer)) + .map_err(eckit_sys::Error::from)?; + Ok(Self::from_raw(inner)) + } + /// Open for reading. Returns the opened handle and estimated data length. pub fn open_for_read(mut self) -> Result<(DataHandle, i64)> { let mut inner = self diff --git a/rust/crates/eckit/tests/datahandle.rs b/rust/crates/eckit/tests/datahandle.rs new file mode 100644 index 000000000..f5ac49a85 --- /dev/null +++ b/rust/crates/eckit/tests/datahandle.rs @@ -0,0 +1,78 @@ +// SPDX-FileCopyrightText: 1996- European Centre for Medium-Range Weather Forecasts (ECMWF) +// SPDX-License-Identifier: Apache-2.0 + +//! `DataHandle` adapters: `from_writer`. + +use std::io::Write; +use std::sync::{Arc, Mutex}; + +use eckit::DataHandle; + +#[derive(Clone)] +struct SharedSink(Arc>>); + +impl Write for SharedSink { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().expect("sink lock").extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +struct FailingSink; + +impl Write for FailingSink { + fn write(&mut self, _buf: &[u8]) -> std::io::Result { + Err(std::io::Error::other("sink failure")) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[test] +fn from_writer_streams_to_rust_sink() -> eckit::Result<()> { + let sink = SharedSink(Arc::new(Mutex::new(Vec::new()))); + let bytes = sink.0.clone(); + + let handle = DataHandle::from_writer(sink)?; + assert!(!handle.can_seek()?); + + let mut handle = handle.open_for_write(0)?; + handle.write_all(b"hello ").expect("write"); + handle.write_all(b"world").expect("write"); + handle.close()?; + + assert_eq!(bytes.lock().expect("sink lock").as_slice(), b"hello world"); + Ok(()) +} + +#[test] +fn from_writer_surfaces_sink_errors() -> eckit::Result<()> { + let mut handle = DataHandle::from_writer(FailingSink)?.open_for_write(0)?; + assert!(handle.write_all(b"data").is_err()); + Ok(()) +} + +#[test] +fn from_writer_surfaces_sink_panics_as_errors() -> eckit::Result<()> { + struct PanickingSink; + + impl Write for PanickingSink { + fn write(&mut self, _buf: &[u8]) -> std::io::Result { + panic!("sink panic"); + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + let mut handle = DataHandle::from_writer(PanickingSink)?.open_for_write(0)?; + assert!(handle.write_all(b"data").is_err()); + Ok(()) +}