feat(rt/tokio): additive tokio and hyper i/o adaptors

src/rt/tokio.rs

@@ -1,4 +1,56 @@
-//! Tokio IO integration for hyper
+//! [`tokio`] runtime components integration for [`hyper`].
+//!
+//! [`hyper::rt`] exposes a set of traits to allow hyper to be agnostic to
+//! its underlying asynchronous runtime. This submodule provides glue for
+//! [`tokio`] users to bridge those types to [`hyper`]'s interfaces.
+//!
+//! # IO
+//!
+//! [`hyper`] abstracts over asynchronous readers and writers using [`Read`]
+//! and [`Write`], while [`tokio`] abstracts over this using [`AsyncRead`]
+//! and [`AsyncWrite`]. This submodule provides a collection of IO adaptors
+//! to bridge these two IO ecosystems together: [`TokioIo<I>`],
+//! [`WithHyperIo<I>`], and [`WithTokioIo<I>`].
+//!
+//! To compare and constrast these IO adaptors and to help explain which
+//! is the proper choice for your needs, here is a table showing which IO
+//! traits these implement, given two types `T` and `H` which implement
+//! Tokio's and Hyper's corresponding IO traits:
+//!
+//! |                    | [`AsyncRead`]    | [`AsyncWrite`]    | [`Read`]     | [`Write`]    |
+//! |--------------------|------------------|-------------------|--------------|--------------|
+//! | `T`                | ✅ **true**      | ✅ **true**       | ❌ **false** | ❌ **false** |
+//! | `H`                | ❌ **false**     | ❌ **false**      | ✅ **true**  | ✅ **true**  |
+//! | [`TokioIo<T>`]     | ❌ **false**     | ❌ **false**      | ✅ **true**  | ✅ **true**  |
+//! | [`TokioIo<H>`]     | ✅ **true**      | ✅ **true**       | ❌ **false** | ❌ **false** |
+//! | [`WithHyperIo<T>`] | ✅ **true**      | ✅ **true**       | ✅ **true**  | ✅ **true**  |
+//! | [`WithHyperIo<H>`] | ❌ **false**     | ❌ **false**      | ❌ **false** | ❌ **false** |
+//! | [`WithTokioIo<T>`] | ❌ **false**     | ❌ **false**      | ❌ **false** | ❌ **false** |
+//! | [`WithTokioIo<H>`] | ✅ **true**      | ✅ **true**       | ✅ **true**  | ✅ **true**  |
+//!
+//! For most situations, [`TokioIo<I>`] is the proper choice. This should be
+//! constructed, wrapping some underlying [`hyper`] or [`tokio`] IO, at the
+//! call-site of a function like [`hyper::client::conn::http1::handshake`].
+//!
+//! [`TokioIo<I>`] switches across these ecosystems, but notably does not
+//! preserve the existing IO trait implementations of its underlying IO. If
+//! one wishes to _extend_ IO with additional implementations,
+//! [`WithHyperIo<I>`] and [`WithTokioIo<I>`] are the correct choice.
+//!
+//! For example, a Tokio reader/writer can be wrapped in [`WithHyperIo<I>`].
+//! That will implement _both_ sets of IO traits. Conversely,
+//! [`WithTokioIo<I>`] will implement both sets of IO traits given a
+//! reader/writer that implements Hyper's [`Read`] and [`Write`].
+//!
+//! See [`tokio::io`] and ["_Asynchronous IO_"][tokio-async-docs] for more
+//! information.
+//!
+//! [`AsyncRead`]: tokio::io::AsyncRead
+//! [`AsyncWrite`]: tokio::io::AsyncWrite
+//! [`Read`]: hyper::rt::Read
+//! [`Write`]: hyper::rt::Write
+//! [tokio-async-docs]: https://docs.rs/tokio/latest/tokio/#asynchronous-io
+
 use std::{
     future::Future,
     pin::Pin,
@@ -12,6 +64,11 @@ use pin_project_lite::pin_project;
 #[cfg(feature = "tracing")]
 use tracing::instrument::Instrument;
 
+pub use self::{with_hyper_io::WithHyperIo, with_tokio_io::WithTokioIo};
+
+mod with_hyper_io;
+mod with_tokio_io;
+
 /// Future executor that utilises `tokio` threads.
 #[non_exhaustive]
 #[derive(Default, Debug, Clone)]

src/rt/tokio/with_hyper_io.rs

@@ -0,0 +1,170 @@
+use pin_project_lite::pin_project;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+pin_project! {
+    /// Extends an underlying [`tokio`] I/O with [`hyper`] I/O implementations.
+    ///
+    /// This implements [`Read`] and [`Write`] given an inner type that implements [`AsyncRead`]
+    /// and [`AsyncWrite`], respectively.
+    #[derive(Debug)]
+    pub struct WithHyperIo<I> {
+        #[pin]
+        inner: I,
+    }
+}
+
+// ==== impl WithHyperIo =====
+
+impl<I> WithHyperIo<I> {
+    /// Wraps the inner I/O in an [`WithHyperIo<I>`]
+    pub fn new(inner: I) -> Self {
+        Self { inner }
+    }
+
+    /// Returns a reference to the inner type.
+    pub fn inner(&self) -> &I {
+        &self.inner
+    }
+
+    /// Returns a mutable reference to the inner type.
+    pub fn inner_mut(&mut self) -> &mut I {
+        &mut self.inner
+    }
+
+    /// Consumes this wrapper and returns the inner type.
+    pub fn into_inner(self) -> I {
+        self.inner
+    }
+}
+
+/// [`WithHyperIo<I>`] is [`Read`] if `I` is [`AsyncRead`].
+///
+/// [`AsyncRead`]: tokio::io::AsyncRead
+/// [`Read`]: hyper::rt::Read
+impl<I> hyper::rt::Read for WithHyperIo<I>
+where
+    I: tokio::io::AsyncRead,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        mut buf: hyper::rt::ReadBufCursor<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        let n = unsafe {
+            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
+            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
+                Poll::Ready(Ok(())) => tbuf.filled().len(),
+                other => return other,
+            }
+        };
+
+        unsafe {
+            buf.advance(n);
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+/// [`WithHyperIo<I>`] is [`Write`] if `I` is [`AsyncWrite`].
+///
+/// [`AsyncWrite`]: tokio::io::AsyncWrite
+/// [`Write`]: hyper::rt::Write
+impl<I> hyper::rt::Write for WithHyperIo<I>
+where
+    I: tokio::io::AsyncWrite,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        tokio::io::AsyncWrite::is_write_vectored(&self.inner)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
+    }
+}
+
+/// [`WithHyperIo<I>`] exposes its inner `I`'s [`AsyncRead`] implementation.
+///
+/// [`AsyncRead`]: tokio::io::AsyncRead
+impl<I> tokio::io::AsyncRead for WithHyperIo<I>
+where
+    I: tokio::io::AsyncRead,
+{
+    #[inline]
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        self.project().inner.poll_read(cx, buf)
+    }
+}
+
+/// [`WithHyperIo<I>`] exposes its inner `I`'s [`AsyncWrite`] implementation.
+///
+/// [`AsyncWrite`]: tokio::io::AsyncWrite
+impl<I> tokio::io::AsyncWrite for WithHyperIo<I>
+where
+    I: tokio::io::AsyncWrite,
+{
+    #[inline]
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        self.project().inner.poll_write(cx, buf)
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        self.project().inner.poll_flush(cx)
+    }
+
+    #[inline]
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        self.project().inner.poll_shutdown(cx)
+    }
+
+    #[inline]
+    fn is_write_vectored(&self) -> bool {
+        self.inner.is_write_vectored()
+    }
+
+    #[inline]
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        self.project().inner.poll_write_vectored(cx, bufs)
+    }
+}