rch: io channel

A channel for streaming binary data with receive verification.

Author: surban

remoc/src/rch/io/mod.rs

@@ -0,0 +1,181 @@
+//! A channel for streaming binary data with receive verification.
+//!
+//! This channel provides [`AsyncWrite`](tokio::io::AsyncWrite) and
+//! [`AsyncRead`](tokio::io::AsyncRead) implementations for streaming binary data
+//! between endpoints. The receiver can verify that all transmitted data has been
+//! received completely, ensuring data integrity.
+//!
+//! # Modes
+//!
+//! The channel supports two modes depending on whether the total data size is known upfront:
+//!
+//! - **Known size** ([`sized`]): The sender specifies the exact number of bytes to transmit.
+//!   The receiver knows the size in advance via [`Receiver::size`] and verifies that the
+//!   received data matches.
+//!
+//! - **Unknown size** ([`channel`]): The sender can write any amount of data.
+//!   The receiver learns the final size only upon completion and verifies integrity.
+//!
+//! In both cases, the receiver returns an error if the received byte count does not match
+//! the expected size.
+//!
+//! # Completion
+//!
+//! - **Known size**: Calling [`shutdown`](tokio::io::AsyncWriteExt::shutdown) is optional.
+//!   The receiver uses the pre-announced size to determine completion.
+//!   However, [`flush`](tokio::io::AsyncWriteExt::flush) must be called before dropping
+//!   the sender to ensure all pending data is transmitted, as per the standard
+//!   [`AsyncWrite`](tokio::io::AsyncWrite) contract.
+//!
+//! - **Unknown size**: Calling [`shutdown`](tokio::io::AsyncWriteExt::shutdown) is **required**.
+//!   This sends the final byte count to the receiver. If the sender is dropped without calling
+//!   shutdown, the receiver will return an [`UnexpectedEof`](std::io::ErrorKind::UnexpectedEof) error.
+//!
+//! # Remote constraint
+//!
+//! At least one half of this channel must be sent to a remote endpoint.
+//! Using both halves locally will cause operations to block indefinitely.
+//!
+//! # Example: Known size
+//!
+//! Use [`sized`] when the data size is known upfront (e.g., file transfer with known file size).
+//!
+//! ```
+//! use remoc::prelude::*;
+//! use tokio::io::{AsyncWriteExt, AsyncReadExt};
+//!
+//! // This would be run on the client.
+//! async fn client(mut tx: rch::base::Sender<rch::io::Receiver>) {
+//!     let (mut io_tx, io_rx) = rch::io::sized(11);
+//!
+//!     // The sender knows the expected size.
+//!     assert_eq!(io_tx.expected_size(), Some(11));
+//!
+//!     // Send receiver to server.
+//!     tx.send(io_rx).await.unwrap();
+//!
+//!     // Write data.
+//!     io_tx.write_all(b"hello world").await.unwrap();
+//!     io_tx.shutdown().await.unwrap();
+//!     // For sized channels, this would be sufficient:
+//!     // io_tx.flush().await.unwrap();
+//! }
+//!
+//! // This would be run on the server.
+//! async fn server(mut rx: rch::base::Receiver<rch::io::Receiver>) {
+//!     let mut io_rx = rx.recv().await.unwrap().unwrap();
+//!
+//!     // The receiver knows the size in advance.
+//!     assert_eq!(io_rx.size(), Some(11));
+//!
+//!     // Read data.
+//!     let mut buf = Vec::new();
+//!     io_rx.read_to_end(&mut buf).await.unwrap();
+//!     assert_eq!(buf, b"hello world");
+//! }
+//! # tokio_test::block_on(remoc::doctest::client_server(client, server));
+//! ```
+//!
+//! # Example: Unknown size
+//!
+//! Use [`channel`] when the data size is not known upfront (e.g., streaming or compressed data).
+//! **Calling shutdown is required** to signal completion and send the final size to the receiver.
+//!
+//! ```
+//! use remoc::prelude::*;
+//! use tokio::io::{AsyncWriteExt, AsyncReadExt};
+//!
+//! // This would be run on the client.
+//! async fn client(mut tx: rch::base::Sender<rch::io::Receiver>) {
+//!     let (mut io_tx, io_rx) = rch::io::channel();
+//!
+//!     // Size is unknown.
+//!     assert_eq!(io_tx.expected_size(), None);
+//!
+//!     // Send receiver to server.
+//!     tx.send(io_rx).await.unwrap();
+//!
+//!     // Write data in chunks (size determined at runtime).
+//!     io_tx.write_all(b"streaming ").await.unwrap();
+//!     io_tx.write_all(b"data").await.unwrap();
+//!
+//!     // REQUIRED: shutdown sends the final size to the receiver.
+//!     // Without this, the receiver will fail with UnexpectedEof.
+//!     io_tx.shutdown().await.unwrap();
+//! }
+//!
+//! // This would be run on the server.
+//! async fn server(mut rx: rch::base::Receiver<rch::io::Receiver>) {
+//!     let mut io_rx = rx.recv().await.unwrap().unwrap();
+//!
+//!     // Size is unknown until EOF.
+//!     assert_eq!(io_rx.size(), None);
+//!
+//!     // Read all data.
+//!     let mut buf = Vec::new();
+//!     io_rx.read_to_end(&mut buf).await.unwrap();
+//!     assert_eq!(buf, b"streaming data");
+//!
+//!     // After EOF, size becomes known.
+//!     assert_eq!(io_rx.size(), Some(14));
+//! }
+//! # tokio_test::block_on(remoc::doctest::client_server(client, server));
+//! ```
+
+use serde::{Deserialize, Serialize};
+
+use super::{bin, oneshot};
+
+mod receiver;
+mod sender;
+
+pub use receiver::Receiver;
+pub use sender::Sender;
+
+use sender::SizeMode;
+
+/// Internal enum to track size information on the receiver side.
+#[derive(Debug, Serialize, Deserialize)]
+pub(super) enum SizeInfo {
+    /// Size was known at channel creation.
+    Determined(u64),
+    /// Size will be received when sender shuts down.
+    Undetermined(oneshot::Receiver<u64, crate::codec::Default>),
+}
+
+/// Creates a new I/O channel with unknown size.
+///
+/// The sender must call [`AsyncWriteExt::shutdown`](tokio::io::AsyncWriteExt::shutdown)
+/// when done writing to signal completion.
+/// The receiver cannot know the size in advance (returns `None` from [`Receiver::size`]).
+///
+/// Both ends can be sent to remote endpoints.
+pub fn channel() -> (Sender, Receiver) {
+    let (bin_tx, bin_rx) = bin::channel();
+    let (size_tx, size_rx) = oneshot::channel();
+
+    let sender = Sender::new(bin_tx, SizeMode::Unknown(size_tx));
+    let receiver = Receiver::new(bin_rx, SizeInfo::Undetermined(size_rx));
+
+    (sender, receiver)
+}
+
+/// Creates a new I/O channel with known size.
+///
+/// The sender is expected to write exactly `size` bytes.
+/// Attempting to write more will result in an error.
+/// Calling [`flush`](tokio::io::AsyncWriteExt::flush) before dropping ensures all data is sent.
+/// If [`shutdown`](tokio::io::AsyncWriteExt::shutdown) is called, it verifies that exactly
+/// `size` bytes were written.
+///
+/// The receiver can query the size via [`Receiver::size`], which returns `Some(size)`.
+///
+/// Both ends can be sent to remote endpoints.
+pub fn sized(size: u64) -> (Sender, Receiver) {
+    let (bin_tx, bin_rx) = bin::channel();
+
+    let sender = Sender::new(bin_tx, SizeMode::Known(size));
+    let receiver = Receiver::new(bin_rx, SizeInfo::Determined(size));
+
+    (sender, receiver)
+}