tor-async-utils: Introduce JoinReadWrite

I wanted one of these for a test stream.  Unaccountably neither
the futures crate, nor tokio, seem to have one!
This commit is contained in:
Ian Jackson 2023-04-05 11:44:06 +01:00
parent 094c9ee085
commit 10f19eaf56
3 changed files with 96 additions and 1 deletions

View File

@ -21,5 +21,5 @@ void = "1"
[dev-dependencies]
futures-await-test = "0.3.0"
tokio = { version = "1.7", features = ["macros", "rt", "rt-multi-thread", "time"] }
tokio = { version = "1.7", features = ["macros", "net", "rt", "rt-multi-thread", "time"] }

View File

@ -0,0 +1,92 @@
//! Join a readable and writeable into a single `AsyncRead` + `AsyncWrite`
//!
//! In some sense this is the converse of
//! [`AsyncReadExt::split`](futures::AsyncReadExt::split).
use std::io::Error;
use std::pin::Pin;
use std::task::{Context, Poll};
use futures::{AsyncRead, AsyncWrite};
use pin_project::pin_project;
/// Async readable/writeable that dispatches reads to `R` and writes to `W`
///
/// `AsyncRead` is forwarded to `R`.
///
/// `AsyncWrite` is forwarded to `W`.
///
/// Does *not* implement any kind of flushing behaviour when switching between reading and writing.
#[pin_project]
pub struct JoinReadWrite<R: AsyncRead, W: AsyncWrite> {
/// readable
#[pin]
r: R,
/// writeable
#[pin]
w: W,
}
impl<R: AsyncRead, W: AsyncWrite> JoinReadWrite<R, W> {
/// Join an `AsyncRead` and an `AsyncWrite` into a single `impl AsyncRead + AsyncWrite`
///
/// ```
/// # #[tokio::main]
/// # async fn main() {
/// use tor_async_utils::JoinReadWrite;
/// use futures::{AsyncReadExt as _, AsyncWriteExt as _};
///
/// let read = b"hello\n";
/// let mut read = &read[..];
/// let mut write = Vec::<u8>::new();
///
/// let mut joined = JoinReadWrite::new(read, write);
///
/// let mut got = String::new();
/// let _: usize = joined.read_to_string(&mut got).await.unwrap();
/// assert_eq!(got, "hello\n");
///
/// let () = joined.write_all(b"some data").await.unwrap();
///
/// let (r, w) = joined.into_parts();
/// assert_eq!(w, b"some data");
/// # }
/// ```
pub fn new(r: R, w: W) -> Self {
JoinReadWrite { r, w }
}
/// Dismantle a `JoinReadWrite` into its constituent `AsyncRead` and `AsyncWrite`
pub fn into_parts(self) -> (R, W) {
let JoinReadWrite { r, w } = self;
(r, w)
}
}
impl<R: AsyncRead, W: AsyncWrite> AsyncRead for JoinReadWrite<R, W> {
fn poll_read(
self: Pin<&mut Self>,
c: &mut Context,
out: &mut [u8],
) -> Poll<Result<usize, Error>> {
self.project().r.poll_read(c, out)
}
}
impl<R: AsyncRead, W: AsyncWrite> AsyncWrite for JoinReadWrite<R, W> {
fn poll_write(
self: Pin<&mut Self>,
c: &mut Context,
data: &[u8],
) -> Poll<Result<usize, Error>> {
self.project().w.poll_write(c, data)
}
fn poll_flush(self: Pin<&mut Self>, c: &mut Context) -> Poll<Result<(), Error>> {
self.project().w.poll_flush(c)
}
fn poll_close(self: Pin<&mut Self>, c: &mut Context) -> Poll<Result<(), Error>> {
self.project().w.poll_close(c)
}
}

View File

@ -38,9 +38,12 @@
#![allow(clippy::result_large_err)] // temporary workaround for arti#587
//! <!-- @@ end lint list maintained by maint/add_warning @@ -->
mod join_read_write;
mod sinkext;
mod watch;
pub use join_read_write::*;
pub use sinkext::{SinkExt, SinkPrepareSendFuture, SinkSendable};
pub use watch::{DropNotifyEofSignallable, DropNotifyWatchSender, PostageWatchSenderExt};