diff --git a/tokio-util/src/io/mod.rs b/tokio-util/src/io/mod.rs index eec74448890..0444a7421ec 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::{ReadBridge, WriteBridge}; 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..73ee2363466 --- /dev/null +++ b/tokio-util/src/io/sync_bridge.rs @@ -0,0 +1,58 @@ +use std::io::{Read, Write}; +use std::pin::Pin; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +/// A [`std::io::Read`] implementation backed by an asynchronous source. +#[derive(Debug)] +pub struct ReadBridge { + reader: Pin>, + rt: tokio::runtime::Handle, +} + +impl Read for ReadBridge { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + let reader = &mut self.reader; + self.rt.block_on(async { reader.read(buf).await }) + } +} + +impl ReadBridge { + /// Create a [`std::io::Read`] implementation backed by an asynchronous source. + /// + /// This is useful with e.g. [`tokio::task::spawn_blocking`]. + pub fn new(reader: T) -> Self { + let reader = Box::pin(reader); + let rt = tokio::runtime::Handle::current(); + ReadBridge { reader, rt } + } +} + +/// A [`std::io::Write`] implementation backed by an asynchronous source. +#[derive(Debug)] +pub struct WriteBridge { + w: Pin>, + rt: tokio::runtime::Handle, +} + +impl Write for WriteBridge { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let w = &mut self.w; + self.rt.block_on(async { w.write(buf).await }) + } + + fn flush(&mut self) -> std::io::Result<()> { + let w = &mut self.w; + self.rt.block_on(async { w.flush().await }) + } +} + +impl WriteBridge { + /// Create a [`std::io::Write`] implementation backed by an asynchronous source. + /// + /// This is useful with e.g. [`tokio::task::spawn_blocking`]. + pub fn new(reader: T) -> Self { + let w = Box::pin(reader); + let rt = tokio::runtime::Handle::current(); + WriteBridge { w, 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..1df1d0aab95 --- /dev/null +++ b/tokio-util/tests/io_sync_bridge.rs @@ -0,0 +1,26 @@ +use std::io::{Cursor, Read, Result as IoResult}; +use tokio::io::AsyncRead; +use tokio_util::io::ReadBridge; + +async fn test_reader_len( + r: impl AsyncRead + Unpin + Send + 'static, + expected_len: usize, +) -> IoResult<()> { + let mut r = ReadBridge::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(()) +}