Add DataHandle::from_writer wrapping Rust Write sinks

rust/crates/eckit-sys/cpp/DataHandleWrapper.cc

@@ -153,6 +153,55 @@ class RustReaderHandle : public eckit::DataHandle {
     rust::Box<ReaderBox> reader_;
 };
 
+// `eckit::DataHandle` subclass that forwards `write()` to a Rust `Write`
+// sink held in a `rust::Box<WriterBox>`. `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<WriterBox> 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<const uint8_t*>(buffer);
+        rust::Slice<const uint8_t> slice{bytes, static_cast<size_t>(length)};
+        int64_t n = invoke_writer_write(*writer_, slice);
+        if (n < 0) {
+            throw eckit::WriteError("RustWriterHandle: error writing to Rust sink");
+        }
+        return static_cast<long>(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<WriterBox> writer_;
+};
+
 }  // namespace
 
 //----------------------------------------------------------------------------------------------------------------------
@@ -161,6 +210,13 @@ std::unique_ptr<DataHandleWrapper> DataHandleWrapper::from_reader(rust::Box<Read
     return std::make_unique<DataHandleWrapper>(new RustReaderHandle(std::move(reader)));
 }
 
+std::unique_ptr<DataHandleWrapper> DataHandleWrapper::from_writer(rust::Box<WriterBox> writer) {
+    auto handle  = std::make_unique<RustWriterHandle>(std::move(writer));
+    auto wrapper = std::make_unique<DataHandleWrapper>(handle.get());
+    handle.release();
+    return wrapper;
+}
+
 //----------------------------------------------------------------------------------------------------------------------
 
 }  // namespace eckit_bridge

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<DataHandleWrapper> from_reader(rust::Box<ReaderBox> 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<DataHandleWrapper> from_writer(rust::Box<WriterBox> writer);
 };
 
 //----------------------------------------------------------------------------------------------------------------------

rust/crates/eckit-sys/src/lib.rs

@@ -159,6 +159,11 @@ mod ffi {
         #[Self = "DataHandleWrapper"]
         fn from_reader(reader: Box<ReaderBox>) -> Result<UniquePtr<DataHandleWrapper>>;
 
+        /// 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<WriterBox>) -> Result<UniquePtr<DataHandleWrapper>>;
+
         // ==================== 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<WriterBox>` 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<dyn Write + Send>`.
+///
+/// The C++ `RustWriterHandle` (declared in `DataHandleWrapper.h` as `struct
+/// WriterBox`) carries this by `rust::Box<WriterBox>` and forwards each C++
+/// `write(const void*, long)` call via `invoke_writer_write`.
+pub struct WriterBox(Box<dyn std::io::Write + Send>);
+
+/// Wrap a Rust `Write` sink for `ffi::from_writer`.
+pub fn make_writer_box<W>(writer: W) -> Box<WriterBox>
+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 {

rust/crates/eckit/src/datahandle.rs

@@ -134,6 +134,21 @@ impl DataHandle<Closed> {
         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<W>(writer: W) -> Result<Self>
+    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<Reading>, i64)> {
         let mut inner = self

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<Mutex<Vec<u8>>>);
+
+impl Write for SharedSink {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        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<usize> {
+        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<usize> {
+            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(())
+}