cyper_core/
stream.rs

1use std::{
2    borrow::Cow,
3    io,
4    mem::MaybeUninit,
5    pin::Pin,
6    task::{Context, Poll, ready},
7};
8
9use compio::{
10    io::{AsyncRead, AsyncWrite, util::Splittable},
11    tls::{MaybeTlsStream, TlsStream},
12};
13use send_wrapper::SendWrapper;
14
15/// A stream wrapper for hyper.
16#[derive(Debug)]
17pub struct HyperStream<S: Splittable>(SendWrapper<MaybeTlsStream<S>>);
18
19impl<S: Splittable> HyperStream<S> {
20    /// Create a new [`HyperStream`] from a plain stream.
21    pub fn new_plain(s: S) -> Self {
22        Self(SendWrapper::new(MaybeTlsStream::new_plain(s)))
23    }
24
25    /// Create a new [`HyperStream`] from a TLS stream.
26    pub fn new_tls(s: TlsStream<S>) -> Self {
27        Self(SendWrapper::new(MaybeTlsStream::new_tls(s)))
28    }
29
30    /// Whether the stream is TLS-encrypted.
31    pub fn is_tls(&self) -> bool {
32        self.0.is_tls()
33    }
34}
35
36impl<S: Splittable + 'static> HyperStream<S>
37where
38    S::ReadHalf: AsyncRead + Unpin,
39    S::WriteHalf: AsyncWrite + Unpin,
40{
41    /// Returns the negotiated ALPN protocol.
42    pub fn negotiated_alpn(&self) -> Option<Cow<'_, [u8]>> {
43        self.0.negotiated_alpn()
44    }
45}
46
47impl<S: Splittable + 'static> hyper::rt::Read for HyperStream<S>
48where
49    S::ReadHalf: AsyncRead + Unpin,
50    S::WriteHalf: AsyncWrite + Unpin,
51{
52    fn poll_read(
53        mut self: Pin<&mut Self>,
54        cx: &mut Context<'_>,
55        mut buf: hyper::rt::ReadBufCursor<'_>,
56    ) -> Poll<io::Result<()>> {
57        let uninit = unsafe { buf.as_mut() };
58        uninit.fill(MaybeUninit::new(0));
59        let res = ready!(futures_util::AsyncRead::poll_read(
60            Pin::new(&mut *self.0),
61            cx,
62            unsafe { uninit.assume_init_mut() }
63        ))?;
64        unsafe { buf.advance(res) };
65        Poll::Ready(Ok(()))
66    }
67}
68
69impl<S: Splittable + 'static> futures_util::AsyncRead for HyperStream<S>
70where
71    S::ReadHalf: AsyncRead + Unpin,
72    S::WriteHalf: AsyncWrite + Unpin,
73{
74    fn poll_read(
75        mut self: Pin<&mut Self>,
76        cx: &mut Context<'_>,
77        buf: &mut [u8],
78    ) -> Poll<io::Result<usize>> {
79        futures_util::AsyncRead::poll_read(Pin::new(&mut *self.0), cx, buf)
80    }
81
82    fn poll_read_vectored(
83        mut self: Pin<&mut Self>,
84        cx: &mut Context<'_>,
85        bufs: &mut [io::IoSliceMut<'_>],
86    ) -> Poll<io::Result<usize>> {
87        futures_util::AsyncRead::poll_read_vectored(Pin::new(&mut *self.0), cx, bufs)
88    }
89}
90
91impl<S: Splittable + 'static> hyper::rt::Write for HyperStream<S>
92where
93    S::ReadHalf: AsyncRead + Unpin,
94    S::WriteHalf: AsyncWrite + Unpin,
95{
96    fn poll_write(
97        mut self: Pin<&mut Self>,
98        cx: &mut Context<'_>,
99        buf: &[u8],
100    ) -> Poll<io::Result<usize>> {
101        futures_util::AsyncWrite::poll_write(Pin::new(&mut *self.0), cx, buf)
102    }
103
104    fn poll_write_vectored(
105        mut self: Pin<&mut Self>,
106        cx: &mut Context<'_>,
107        bufs: &[io::IoSlice<'_>],
108    ) -> Poll<io::Result<usize>> {
109        futures_util::AsyncWrite::poll_write_vectored(Pin::new(&mut *self.0), cx, bufs)
110    }
111
112    fn is_write_vectored(&self) -> bool {
113        true
114    }
115
116    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
117        futures_util::AsyncWrite::poll_flush(Pin::new(&mut *self.0), cx)
118    }
119
120    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
121        futures_util::AsyncWrite::poll_close(Pin::new(&mut *self.0), cx)
122    }
123}
124
125impl<S: Splittable + 'static> futures_util::AsyncWrite for HyperStream<S>
126where
127    S::ReadHalf: AsyncRead + Unpin,
128    S::WriteHalf: AsyncWrite + Unpin,
129{
130    fn poll_write(
131        mut self: Pin<&mut Self>,
132        cx: &mut Context<'_>,
133        buf: &[u8],
134    ) -> Poll<io::Result<usize>> {
135        futures_util::AsyncWrite::poll_write(Pin::new(&mut *self.0), cx, buf)
136    }
137
138    fn poll_write_vectored(
139        mut self: Pin<&mut Self>,
140        cx: &mut Context<'_>,
141        bufs: &[io::IoSlice<'_>],
142    ) -> Poll<io::Result<usize>> {
143        futures_util::AsyncWrite::poll_write_vectored(Pin::new(&mut *self.0), cx, bufs)
144    }
145
146    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
147        futures_util::AsyncWrite::poll_flush(Pin::new(&mut *self.0), cx)
148    }
149
150    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
151        futures_util::AsyncWrite::poll_close(Pin::new(&mut *self.0), cx)
152    }
153}
cyper_core/stream.rs

cyper_core/
stream.rs
