diff --git a/tokio-util/src/io/mod.rs b/tokio-util/src/io/mod.rs index eec74448890..54d14b328e5 100644 --- a/tokio-util/src/io/mod.rs +++ b/tokio-util/src/io/mod.rs @@ -1,16 +1,22 @@ //! Helpers for IO related tasks. //! -//! These types are often used in combination with hyper or reqwest, as they +//! The stream types are often used in combination with hyper or reqwest, as they //! allow converting between a hyper [`Body`] and [`AsyncRead`]. //! +//! The [`ReadBridge`] and [`WriteBridge`] convert from the world of async I/O +//! to synchronous I/O; this may often come up when using synchronous APIs +//! inside [`tokio::task::spawn_blocking`]. +//! //! [`Body`]: https://docs.rs/hyper/0.13/hyper/struct.Body.html //! [`AsyncRead`]: tokio::io::AsyncRead mod read_buf; mod reader_stream; mod stream_reader; +mod sync_bridge; pub use self::read_buf::read_buf; pub use self::reader_stream::ReaderStream; pub use self::stream_reader::StreamReader; +pub use self::sync_bridge::SyncIOBridge; pub use crate::util::{poll_read_buf, poll_write_buf}; diff --git a/tokio-util/src/io/sync_bridge.rs b/tokio-util/src/io/sync_bridge.rs new file mode 100644 index 00000000000..f6e1fc7dff1 --- /dev/null +++ b/tokio-util/src/io/sync_bridge.rs @@ -0,0 +1,46 @@ +use pin_project_lite::pin_project; +use std::io::{Read, Write}; +use std::pin::Pin; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +pin_project! { +/// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or +/// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`]. +#[derive(Debug)] +pub struct SyncIOBridge { + src: Pin>, + #[pin] + rt: tokio::runtime::Handle, +} +} + +impl Read for SyncIOBridge { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + let src = &mut self.src; + self.rt.block_on(src.read(buf)) + } +} + +impl Write for SyncIOBridge { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let src = &mut self.src; + self.rt.block_on(src.write(buf)) + } + + fn flush(&mut self) -> std::io::Result<()> { + let src = &mut self.src; + self.rt.block_on(src.flush()) + } +} + +impl SyncIOBridge { + /// Use a [`tokio::io::AsyncRead`] synchronously as a [`std::io::Read`] or + /// a [`tokio::io::AsyncWrite`] as a [`std::io::Write`]. + /// + /// This is useful with e.g. [`tokio::task::spawn_blocking`]. + pub fn new(src: T) -> Self { + let src = Box::pin(src); + let rt = tokio::runtime::Handle::current(); + Self { src, rt } + } +} diff --git a/tokio-util/tests/io_sync_bridge.rs b/tokio-util/tests/io_sync_bridge.rs new file mode 100644 index 00000000000..a5e7bb3569a --- /dev/null +++ b/tokio-util/tests/io_sync_bridge.rs @@ -0,0 +1,43 @@ +#![cfg(feature = "io")] + +use std::error::Error; +use std::io::{Cursor, Read, Result as IoResult}; +use tokio::io::AsyncRead; +use tokio_util::io::SyncIOBridge; + +async fn test_reader_len( + r: impl AsyncRead + Unpin + Send + 'static, + expected_len: usize, +) -> IoResult<()> { + let mut r = SyncIOBridge::new(r); + let res = tokio::task::spawn_blocking(move || { + let mut buf = Vec::new(); + r.read_to_end(&mut buf)?; + Ok::<_, std::io::Error>(buf) + }) + .await?; + assert_eq!(res?.len(), expected_len); + Ok(()) +} + +#[tokio::test] +async fn test_async_read_to_sync() -> Result<(), Box> { + test_reader_len(tokio::io::empty(), 0).await?; + let buf = b"hello world"; + test_reader_len(Cursor::new(buf), buf.len()).await?; + Ok(()) +} + +#[tokio::test] +async fn test_async_write_to_sync() -> Result<(), Box> { + let mut dest = Vec::new(); + let src = b"hello world"; + let dest = tokio::task::spawn_blocking(move || -> Result<_, String> { + let mut w = SyncIOBridge::new(Cursor::new(&mut dest)); + std::io::copy(&mut Cursor::new(src), &mut w).map_err(|e| e.to_string())?; + Ok(dest) + }) + .await??; + assert_eq!(dest.as_slice(), src); + Ok(()) +}