hyper_tokio_io/
lib.rs

1use std::{
2    pin::Pin,
3    task::{Context, Poll},
4};
5
6use pin_project_lite::pin_project;
7
8pin_project! {
9    /// A wrapper that implements Tokio's IO traits for an inner type that
10    /// implements hyper's IO traits, or vice versa (implements hyper's IO
11    /// traits for a type that implements Tokio's IO traits).
12    #[derive(Debug)]
13    pub struct TokioIo<T> {
14        #[pin]
15        inner: T,
16    }
17}
18
19impl<T> TokioIo<T> {
20    /// Wrap a type implementing Tokio's or hyper's IO traits.
21    pub fn new(inner: T) -> Self {
22        Self { inner }
23    }
24
25    /// Borrow the inner type.
26    pub fn inner(&self) -> &T {
27        &self.inner
28    }
29
30    /// Mut borrow the inner type.
31    pub fn inner_mut(&mut self) -> &mut T {
32        &mut self.inner
33    }
34
35    /// Consume this wrapper and get the inner type.
36    pub fn into_inner(self) -> T {
37        self.inner
38    }
39}
40
41impl<T> hyper::rt::Read for TokioIo<T>
42where
43    T: tokio::io::AsyncRead,
44{
45    fn poll_read(
46        self: Pin<&mut Self>,
47        cx: &mut Context<'_>,
48        mut buf: hyper::rt::ReadBufCursor<'_>,
49    ) -> Poll<Result<(), std::io::Error>> {
50        let n = unsafe {
51            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
52            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
53                Poll::Ready(Ok(())) => tbuf.filled().len(),
54                other => return other,
55            }
56        };
57
58        unsafe {
59            buf.advance(n);
60        }
61        Poll::Ready(Ok(()))
62    }
63}
64
65impl<T> hyper::rt::Write for TokioIo<T>
66where
67    T: tokio::io::AsyncWrite,
68{
69    fn poll_write(
70        self: Pin<&mut Self>,
71        cx: &mut Context<'_>,
72        buf: &[u8],
73    ) -> Poll<Result<usize, std::io::Error>> {
74        tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
75    }
76
77    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
78        tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
79    }
80
81    fn poll_shutdown(
82        self: Pin<&mut Self>,
83        cx: &mut Context<'_>,
84    ) -> Poll<Result<(), std::io::Error>> {
85        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
86    }
87
88    fn is_write_vectored(&self) -> bool {
89        tokio::io::AsyncWrite::is_write_vectored(&self.inner)
90    }
91
92    fn poll_write_vectored(
93        self: Pin<&mut Self>,
94        cx: &mut Context<'_>,
95        bufs: &[std::io::IoSlice<'_>],
96    ) -> Poll<Result<usize, std::io::Error>> {
97        tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
98    }
99}
100
101impl<T> tokio::io::AsyncRead for TokioIo<T>
102where
103    T: hyper::rt::Read,
104{
105    fn poll_read(
106        self: Pin<&mut Self>,
107        cx: &mut Context<'_>,
108        tbuf: &mut tokio::io::ReadBuf<'_>,
109    ) -> Poll<Result<(), std::io::Error>> {
110        //let init = tbuf.initialized().len();
111        let filled = tbuf.filled().len();
112        let sub_filled = unsafe {
113            let mut buf = hyper::rt::ReadBuf::uninit(tbuf.unfilled_mut());
114
115            match hyper::rt::Read::poll_read(self.project().inner, cx, buf.unfilled()) {
116                Poll::Ready(Ok(())) => buf.filled().len(),
117                other => return other,
118            }
119        };
120
121        let n_filled = filled + sub_filled;
122        // At least sub_filled bytes had to have been initialized.
123        let n_init = sub_filled;
124        unsafe {
125            tbuf.assume_init(n_init);
126            tbuf.set_filled(n_filled);
127        }
128
129        Poll::Ready(Ok(()))
130    }
131}
132
133impl<T> tokio::io::AsyncWrite for TokioIo<T>
134where
135    T: hyper::rt::Write,
136{
137    fn poll_write(
138        self: Pin<&mut Self>,
139        cx: &mut Context<'_>,
140        buf: &[u8],
141    ) -> Poll<Result<usize, std::io::Error>> {
142        hyper::rt::Write::poll_write(self.project().inner, cx, buf)
143    }
144
145    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
146        hyper::rt::Write::poll_flush(self.project().inner, cx)
147    }
148
149    fn poll_shutdown(
150        self: Pin<&mut Self>,
151        cx: &mut Context<'_>,
152    ) -> Poll<Result<(), std::io::Error>> {
153        hyper::rt::Write::poll_shutdown(self.project().inner, cx)
154    }
155
156    fn is_write_vectored(&self) -> bool {
157        hyper::rt::Write::is_write_vectored(&self.inner)
158    }
159
160    fn poll_write_vectored(
161        self: Pin<&mut Self>,
162        cx: &mut Context<'_>,
163        bufs: &[std::io::IoSlice<'_>],
164    ) -> Poll<Result<usize, std::io::Error>> {
165        hyper::rt::Write::poll_write_vectored(self.project().inner, cx, bufs)
166    }
167}
hyper_tokio_io/lib.rs

hyper_tokio_io/
lib.rs
