hyper/common/io/
compat.rs

1use std::pin::Pin;
2use std::task::{Context, Poll};
3
4/// This adapts from `hyper` IO traits to the ones in Tokio.
5///
6/// This is currently used by `h2`, and by hyper internal unit tests.
7#[derive(Debug)]
8pub(crate) struct Compat<T>(pub(crate) T);
9
10impl<T> Compat<T> {
11    pub(crate) fn new(io: T) -> Self {
12        Compat(io)
13    }
14
15    fn p(self: Pin<&mut Self>) -> Pin<&mut T> {
16        // SAFETY: The simplest of projections. This is just
17        // a wrapper, we don't do anything that would undo the projection.
18        unsafe { self.map_unchecked_mut(|me| &mut me.0) }
19    }
20}
21
22impl<T> tokio::io::AsyncRead for Compat<T>
23where
24    T: crate::rt::Read,
25{
26    /// `poll_read` fn implementation for `Compat<T>`.
27    fn poll_read(
28        self: Pin<&mut Self>,
29        cx: &mut Context<'_>,
30        tbuf: &mut tokio::io::ReadBuf<'_>,
31    ) -> Poll<Result<(), std::io::Error>> {
32        let init = tbuf.initialized().len();
33        let filled = tbuf.filled().len();
34        // SAFETY:
35        // 1. `tbuf.inner_mut()` returns a raw pointer/mutable slice which we wrap into
36        //    a `crate::rt::ReadBuf` that is layout-compatible with the source `tokio::io::ReadBuf`.
37        // 2. We explicitly restore the `init` and `filled` states from the original buffer
38        //    to maintain the invariant that the new `ReadBuf` tracks the same progress.
39        // 3. The underlying memory remains valid and uniquely accessible via `tbuf` for
40        //    the duration of this poll operation.
41        let (new_init, new_filled) = unsafe {
42            let mut buf = crate::rt::ReadBuf::uninit(tbuf.inner_mut());
43            buf.set_init(init);
44            buf.set_filled(filled);
45
46            match crate::rt::Read::poll_read(self.p(), cx, buf.unfilled()) {
47                Poll::Ready(Ok(())) => (buf.init_len(), buf.len()),
48                other => return other,
49            }
50        };
51
52        let n_init = new_init - init;
53        // SAFETY:
54        // 1. `tbuf.assume_init(n_init)` is safe because `crate::rt::Read::poll_read`
55        //    guarantees that the bytes written into the buffer were initialized.
56        // 2. `tbuf.set_filled(new_filled)` is safe because `new_filled` is derived
57        //    directly from the buffer state after the successful read operation.
58        unsafe {
59            tbuf.assume_init(n_init);
60            tbuf.set_filled(new_filled);
61        }
62
63        Poll::Ready(Ok(()))
64    }
65}
66
67impl<T> tokio::io::AsyncWrite for Compat<T>
68where
69    T: crate::rt::Write,
70{
71    fn poll_write(
72        self: Pin<&mut Self>,
73        cx: &mut Context<'_>,
74        buf: &[u8],
75    ) -> Poll<Result<usize, std::io::Error>> {
76        crate::rt::Write::poll_write(self.p(), cx, buf)
77    }
78
79    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
80        crate::rt::Write::poll_flush(self.p(), cx)
81    }
82
83    fn poll_shutdown(
84        self: Pin<&mut Self>,
85        cx: &mut Context<'_>,
86    ) -> Poll<Result<(), std::io::Error>> {
87        crate::rt::Write::poll_shutdown(self.p(), cx)
88    }
89
90    fn is_write_vectored(&self) -> bool {
91        crate::rt::Write::is_write_vectored(&self.0)
92    }
93
94    fn poll_write_vectored(
95        self: Pin<&mut Self>,
96        cx: &mut Context<'_>,
97        bufs: &[std::io::IoSlice<'_>],
98    ) -> Poll<Result<usize, std::io::Error>> {
99        crate::rt::Write::poll_write_vectored(self.p(), cx, bufs)
100    }
101}
102
103#[cfg(test)]
104impl<T> crate::rt::Read for Compat<T>
105where
106    T: tokio::io::AsyncRead,
107{
108    fn poll_read(
109        self: Pin<&mut Self>,
110        cx: &mut Context<'_>,
111        mut buf: crate::rt::ReadBufCursor<'_>,
112    ) -> Poll<Result<(), std::io::Error>> {
113        let n = unsafe {
114            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
115            match tokio::io::AsyncRead::poll_read(self.p(), cx, &mut tbuf) {
116                Poll::Ready(Ok(())) => tbuf.filled().len(),
117                other => return other,
118            }
119        };
120
121        unsafe {
122            buf.advance(n);
123        }
124        Poll::Ready(Ok(()))
125    }
126}
127
128#[cfg(test)]
129impl<T> crate::rt::Write for Compat<T>
130where
131    T: tokio::io::AsyncWrite,
132{
133    fn poll_write(
134        self: Pin<&mut Self>,
135        cx: &mut Context<'_>,
136        buf: &[u8],
137    ) -> Poll<Result<usize, std::io::Error>> {
138        tokio::io::AsyncWrite::poll_write(self.p(), cx, buf)
139    }
140
141    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
142        tokio::io::AsyncWrite::poll_flush(self.p(), cx)
143    }
144
145    fn poll_shutdown(
146        self: Pin<&mut Self>,
147        cx: &mut Context<'_>,
148    ) -> Poll<Result<(), std::io::Error>> {
149        tokio::io::AsyncWrite::poll_shutdown(self.p(), cx)
150    }
151
152    fn is_write_vectored(&self) -> bool {
153        tokio::io::AsyncWrite::is_write_vectored(&self.0)
154    }
155
156    fn poll_write_vectored(
157        self: Pin<&mut Self>,
158        cx: &mut Context<'_>,
159        bufs: &[std::io::IoSlice<'_>],
160    ) -> Poll<Result<usize, std::io::Error>> {
161        tokio::io::AsyncWrite::poll_write_vectored(self.p(), cx, bufs)
162    }
163}
hyper/common/io/compat.rs

hyper/common/io/
compat.rs
