/rust/registry/src/index.crates.io-1949cf8c6b5b557f/rustls-0.23.43/src/stream.rs

Source
use core::ops::{Deref, DerefMut};
use std::io::{BufRead, IoSlice, Read, Result, Write};

use crate::conn::{ConnectionCommon, SideData};

/// This type implements `io::Read` and `io::Write`, encapsulating
/// a Connection `C` and an underlying transport `T`, such as a socket.
///
/// Relies on [`ConnectionCommon::complete_io()`] to perform the necessary I/O.
///
/// This allows you to use a rustls Connection like a normal stream.
#[derive(Debug)]
pub struct Stream<'a, C: 'a + ?Sized, T: 'a + Read + Write + ?Sized> {
    /// Our TLS connection
    pub conn: &'a mut C,

    /// The underlying transport, like a socket
    pub sock: &'a mut T,
}

impl<'a, C, T, S> Stream<'a, C, T>
where
    C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: 'a + Read + Write,
    S: SideData,
{
    /// Make a new Stream using the Connection `conn` and socket-like object
    /// `sock`.  This does not fail and does no IO.
    pub fn new(conn: &'a mut C, sock: &'a mut T) -> Self {
        Self { conn, sock }
    }

    /// If we're handshaking, complete all the IO for that.
    /// If we have data to write, write it all.
    fn complete_prior_io(&mut self) -> Result<()> {
        if self.conn.is_handshaking() {
            self.conn.complete_io(self.sock)?;
        }

        if self.conn.wants_write() {
            self.conn.complete_io(self.sock)?;
        }

        Ok(())
    }

    fn prepare_read(&mut self) -> Result<()> {
        self.complete_prior_io()?;

        // We call complete_io() in a loop since a single call may read only
        // a partial packet from the underlying transport. A full packet is
        // needed to get more plaintext, which we must do if EOF has not been
        // hit.
        while self.conn.wants_read() {
            if self.conn.complete_io(self.sock)?.0 == 0 {
                break;
            }
        }

        Ok(())
    }

    // Implements `BufRead::fill_buf` but with more flexible lifetimes, so StreamOwned can reuse it
    fn fill_buf(mut self) -> Result<&'a [u8]>
    where
        S: 'a,
    {
        self.prepare_read()?;
        self.conn.reader().into_first_chunk()
    }
}

impl<'a, C, T, S> Read for Stream<'a, C, T>
where
    C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: 'a + Read + Write,
    S: SideData,
{
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        self.prepare_read()?;
        self.conn.reader().read(buf)
    }

    #[cfg(read_buf)]
    fn read_buf(&mut self, cursor: core::io::BorrowedCursor<'_, u8>) -> Result<()> {
        self.prepare_read()?;
        self.conn.reader().read_buf(cursor)
    }
}

impl<'a, C, T, S> BufRead for Stream<'a, C, T>
where
    C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: 'a + Read + Write,
    S: 'a + SideData,
{
    fn fill_buf(&mut self) -> Result<&[u8]> {
        // reborrow to get an owned `Stream`
        Stream {
            conn: self.conn,
            sock: self.sock,
        }
        .fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.conn.reader().consume(amt)
    }
}

impl<'a, C, T, S> Write for Stream<'a, C, T>
where
    C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: 'a + Read + Write,
    S: SideData,
{
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        self.complete_prior_io()?;

        let len = self.conn.writer().write(buf)?;

        // Try to write the underlying transport here, but don't let
        // any errors mask the fact we've consumed `len` bytes.
        // Callers will learn of permanent errors on the next call.
        let _ = self.conn.complete_io(self.sock);

        Ok(len)
    }

    fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
        self.complete_prior_io()?;

        let len = self
            .conn
            .writer()
            .write_vectored(bufs)?;

        // Try to write the underlying transport here, but don't let
        // any errors mask the fact we've consumed `len` bytes.
        // Callers will learn of permanent errors on the next call.
        let _ = self.conn.complete_io(self.sock);

        Ok(len)
    }

    fn flush(&mut self) -> Result<()> {
        self.complete_prior_io()?;

        self.conn.writer().flush()?;
        if self.conn.wants_write() {
            self.conn.complete_io(self.sock)?;
        }
        Ok(())
    }
}

/// This type implements `io::Read` and `io::Write`, encapsulating
/// and owning a Connection `C` and an underlying transport `T`, such as a socket.
///
/// Relies on [`ConnectionCommon::complete_io()`] to perform the necessary I/O.
///
/// This allows you to use a rustls Connection like a normal stream.
#[derive(Debug)]
pub struct StreamOwned<C: Sized, T: Read + Write + Sized> {
    /// Our connection
    pub conn: C,

    /// The underlying transport, like a socket
    pub sock: T,
}

impl<C, T, S> StreamOwned<C, T>
where
    C: DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: Read + Write,
    S: SideData,
{
    /// Make a new StreamOwned taking the Connection `conn` and socket-like
    /// object `sock`.  This does not fail and does no IO.
    ///
    /// This is the same as `Stream::new` except `conn` and `sock` are
    /// moved into the StreamOwned.
    pub fn new(conn: C, sock: T) -> Self {
        Self { conn, sock }
    }

    /// Get a reference to the underlying socket
    pub fn get_ref(&self) -> &T {
        &self.sock
    }

    /// Get a mutable reference to the underlying socket
    pub fn get_mut(&mut self) -> &mut T {
        &mut self.sock
    }

    /// Extract the `conn` and `sock` parts from the `StreamOwned`
    pub fn into_parts(self) -> (C, T) {
        (self.conn, self.sock)
    }
}

impl<'a, C, T, S> StreamOwned<C, T>
where
    C: DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: Read + Write,
    S: SideData,
{
    fn as_stream(&'a mut self) -> Stream<'a, C, T> {
        Stream {
            conn: &mut self.conn,
            sock: &mut self.sock,
        }
    }
}

impl<C, T, S> Read for StreamOwned<C, T>
where
    C: DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: Read + Write,
    S: SideData,
{
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        self.as_stream().read(buf)
    }

    #[cfg(read_buf)]
    fn read_buf(&mut self, cursor: core::io::BorrowedCursor<'_, u8>) -> Result<()> {
        self.as_stream().read_buf(cursor)
    }
}

impl<C, T, S> BufRead for StreamOwned<C, T>
where
    C: DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: Read + Write,
    S: 'static + SideData,
{
    fn fill_buf(&mut self) -> Result<&[u8]> {
        self.as_stream().fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.as_stream().consume(amt)
    }
}

impl<C, T, S> Write for StreamOwned<C, T>
where
    C: DerefMut + Deref<Target = ConnectionCommon<S>>,
    T: Read + Write,
    S: SideData,
{
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        self.as_stream().write(buf)
    }

    fn flush(&mut self) -> Result<()> {
        self.as_stream().flush()
    }
}

#[cfg(test)]
mod tests {
    use std::net::TcpStream;

    use super::{Stream, StreamOwned};
    use crate::client::ClientConnection;
    use crate::server::ServerConnection;

    #[test]
    fn stream_can_be_created_for_connection_and_tcpstream() {
        type _Test<'a> = Stream<'a, ClientConnection, TcpStream>;
    }

    #[test]
    fn streamowned_can_be_created_for_client_and_tcpstream() {
        type _Test = StreamOwned<ClientConnection, TcpStream>;
    }

    #[test]
    fn streamowned_can_be_created_for_server_and_tcpstream() {
        type _Test = StreamOwned<ServerConnection, TcpStream>;
    }
}

Coverage Report

Created: 2026-09-04 06:48

Line	Count	Source
1		use core::ops::{Deref, DerefMut};
2		use std::io::{BufRead, IoSlice, Read, Result, Write};
3
4		use crate::conn::{ConnectionCommon, SideData};
5
6		/// This type implements `io::Read` and `io::Write`, encapsulating
7		/// a Connection `C` and an underlying transport `T`, such as a socket.
8		///
9		/// Relies on [`ConnectionCommon::complete_io()`] to perform the necessary I/O.
10		///
11		/// This allows you to use a rustls Connection like a normal stream.
12		#[derive(Debug)]
13		pub struct Stream<'a, C: 'a + ?Sized, T: 'a + Read + Write + ?Sized> {
14		/// Our TLS connection
15		pub conn: &'a mut C,
16
17		/// The underlying transport, like a socket
18		pub sock: &'a mut T,
19		}
20
21		impl<'a, C, T, S> Stream<'a, C, T>
22		where
23		C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
24		T: 'a + Read + Write,
25		S: SideData,
26		{
27		/// Make a new Stream using the Connection `conn` and socket-like object
28		/// `sock`. This does not fail and does no IO.
29	0	pub fn new(conn: &'a mut C, sock: &'a mut T) -> Self {
30	0	Self { conn, sock }
31	0	}
32
33		/// If we're handshaking, complete all the IO for that.
34		/// If we have data to write, write it all.
35	0	fn complete_prior_io(&mut self) -> Result<()> {
36	0	if self.conn.is_handshaking() {
37	0	self.conn.complete_io(self.sock)?;
38	0	}
39
40	0	if self.conn.wants_write() {
41	0	self.conn.complete_io(self.sock)?;
42	0	}
43
44	0	Ok(())
45	0	} Unexecuted instantiation: <rustls::stream::Stream<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>>>::complete_prior_io Unexecuted instantiation: <rustls::stream::Stream<_, _>>::complete_prior_io
46
47	0	fn prepare_read(&mut self) -> Result<()> {
48	0	self.complete_prior_io()?;
49
50		// We call complete_io() in a loop since a single call may read only
51		// a partial packet from the underlying transport. A full packet is
52		// needed to get more plaintext, which we must do if EOF has not been
53		// hit.
54	0	while self.conn.wants_read() {
55	0	if self.conn.complete_io(self.sock)?.0 == 0 {
56	0	break;
57	0	}
58		}
59
60	0	Ok(())
61	0	} Unexecuted instantiation: <rustls::stream::Stream<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>>>::prepare_read Unexecuted instantiation: <rustls::stream::Stream<_, _>>::prepare_read
62
63		// Implements `BufRead::fill_buf` but with more flexible lifetimes, so StreamOwned can reuse it
64	0	fn fill_buf(mut self) -> Result<&'a [u8]>
65	0	where
66	0	S: 'a,
67		{
68	0	self.prepare_read()?;
69	0	self.conn.reader().into_first_chunk()
70	0	}
71		}
72
73		impl<'a, C, T, S> Read for Stream<'a, C, T>
74		where
75		C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
76		T: 'a + Read + Write,
77		S: SideData,
78		{
79	0	fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
80	0	self.prepare_read()?;
81	0	self.conn.reader().read(buf)
82	0	} Unexecuted instantiation: <rustls::stream::Stream<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>> as std::io::Read>::read Unexecuted instantiation: <rustls::stream::Stream<_, _> as std::io::Read>::read
83
84		#[cfg(read_buf)]
85		fn read_buf(&mut self, cursor: core::io::BorrowedCursor<'_, u8>) -> Result<()> {
86		self.prepare_read()?;
87		self.conn.reader().read_buf(cursor)
88		}
89		}
90
91		impl<'a, C, T, S> BufRead for Stream<'a, C, T>
92		where
93		C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
94		T: 'a + Read + Write,
95		S: 'a + SideData,
96		{
97	0	fn fill_buf(&mut self) -> Result<&[u8]> {
98		// reborrow to get an owned `Stream`
99	0	Stream {
100	0	conn: self.conn,
101	0	sock: self.sock,
102	0	}
103	0	.fill_buf()
104	0	}
105
106	0	fn consume(&mut self, amt: usize) {
107	0	self.conn.reader().consume(amt)
108	0	}
109		}
110
111		impl<'a, C, T, S> Write for Stream<'a, C, T>
112		where
113		C: 'a + DerefMut + Deref<Target = ConnectionCommon<S>>,
114		T: 'a + Read + Write,
115		S: SideData,
116		{
117	0	fn write(&mut self, buf: &[u8]) -> Result<usize> {
118	0	self.complete_prior_io()?;
119
120	0	let len = self.conn.writer().write(buf)?;
121
122		// Try to write the underlying transport here, but don't let
123		// any errors mask the fact we've consumed `len` bytes.
124		// Callers will learn of permanent errors on the next call.
125	0	let _ = self.conn.complete_io(self.sock);
126
127	0	Ok(len)
128	0	} Unexecuted instantiation: <rustls::stream::Stream<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>> as std::io::Write>::write Unexecuted instantiation: <rustls::stream::Stream<_, _> as std::io::Write>::write
129
130	0	fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
131	0	self.complete_prior_io()?;
132
133	0	let len = self
134	0	.conn
135	0	.writer()
136	0	.write_vectored(bufs)?;
137
138		// Try to write the underlying transport here, but don't let
139		// any errors mask the fact we've consumed `len` bytes.
140		// Callers will learn of permanent errors on the next call.
141	0	let _ = self.conn.complete_io(self.sock);
142
143	0	Ok(len)
144	0	}
145
146	0	fn flush(&mut self) -> Result<()> {
147	0	self.complete_prior_io()?;
148
149	0	self.conn.writer().flush()?;
150	0	if self.conn.wants_write() {
151	0	self.conn.complete_io(self.sock)?;
152	0	}
153	0	Ok(())
154	0	}
155		}
156
157		/// This type implements `io::Read` and `io::Write`, encapsulating
158		/// and owning a Connection `C` and an underlying transport `T`, such as a socket.
159		///
160		/// Relies on [`ConnectionCommon::complete_io()`] to perform the necessary I/O.
161		///
162		/// This allows you to use a rustls Connection like a normal stream.
163		#[derive(Debug)]
164		pub struct StreamOwned<C: Sized, T: Read + Write + Sized> {
165		/// Our connection
166		pub conn: C,
167
168		/// The underlying transport, like a socket
169		pub sock: T,
170		}
171
172		impl<C, T, S> StreamOwned<C, T>
173		where
174		C: DerefMut + Deref<Target = ConnectionCommon<S>>,
175		T: Read + Write,
176		S: SideData,
177		{
178		/// Make a new StreamOwned taking the Connection `conn` and socket-like
179		/// object `sock`. This does not fail and does no IO.
180		///
181		/// This is the same as `Stream::new` except `conn` and `sock` are
182		/// moved into the StreamOwned.
183	0	pub fn new(conn: C, sock: T) -> Self {
184	0	Self { conn, sock }
185	0	}
186
187		/// Get a reference to the underlying socket
188	0	pub fn get_ref(&self) -> &T {
189	0	&self.sock
190	0	}
191
192		/// Get a mutable reference to the underlying socket
193	0	pub fn get_mut(&mut self) -> &mut T {
194	0	&mut self.sock
195	0	} Unexecuted instantiation: <rustls::stream::StreamOwned<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>>>::get_mut Unexecuted instantiation: <rustls::stream::StreamOwned<_, _>>::get_mut
196
197		/// Extract the `conn` and `sock` parts from the `StreamOwned`
198	0	pub fn into_parts(self) -> (C, T) {
199	0	(self.conn, self.sock)
200	0	}
201		}
202
203		impl<'a, C, T, S> StreamOwned<C, T>
204		where
205		C: DerefMut + Deref<Target = ConnectionCommon<S>>,
206		T: Read + Write,
207		S: SideData,
208		{
209	0	fn as_stream(&'a mut self) -> Stream<'a, C, T> {
210	0	Stream {
211	0	conn: &mut self.conn,
212	0	sock: &mut self.sock,
213	0	}
214	0	} Unexecuted instantiation: <rustls::stream::StreamOwned<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>>>::as_stream Unexecuted instantiation: <rustls::stream::StreamOwned<_, _>>::as_stream
215		}
216
217		impl<C, T, S> Read for StreamOwned<C, T>
218		where
219		C: DerefMut + Deref<Target = ConnectionCommon<S>>,
220		T: Read + Write,
221		S: SideData,
222		{
223	0	fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
224	0	self.as_stream().read(buf)
225	0	} Unexecuted instantiation: <rustls::stream::StreamOwned<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>> as std::io::Read>::read Unexecuted instantiation: <rustls::stream::StreamOwned<_, _> as std::io::Read>::read
226
227		#[cfg(read_buf)]
228		fn read_buf(&mut self, cursor: core::io::BorrowedCursor<'_, u8>) -> Result<()> {
229		self.as_stream().read_buf(cursor)
230		}
231		}
232
233		impl<C, T, S> BufRead for StreamOwned<C, T>
234		where
235		C: DerefMut + Deref<Target = ConnectionCommon<S>>,
236		T: Read + Write,
237		S: 'static + SideData,
238		{
239	0	fn fill_buf(&mut self) -> Result<&[u8]> {
240	0	self.as_stream().fill_buf()
241	0	}
242
243	0	fn consume(&mut self, amt: usize) {
244	0	self.as_stream().consume(amt)
245	0	}
246		}
247
248		impl<C, T, S> Write for StreamOwned<C, T>
249		where
250		C: DerefMut + Deref<Target = ConnectionCommon<S>>,
251		T: Read + Write,
252		S: SideData,
253		{
254	0	fn write(&mut self, buf: &[u8]) -> Result<usize> {
255	0	self.as_stream().write(buf)
256	0	} Unexecuted instantiation: <rustls::stream::StreamOwned<rustls::client::client_conn::connection::ClientConnection, ureq::unversioned::transport::io::TransportAdapter<alloc::boxed::Box<dyn ureq::unversioned::transport::Transport>>> as std::io::Write>::write Unexecuted instantiation: <rustls::stream::StreamOwned<_, _> as std::io::Write>::write
257
258	0	fn flush(&mut self) -> Result<()> {
259	0	self.as_stream().flush()
260	0	}
261		}
262
263		#[cfg(test)]
264		mod tests {
265		use std::net::TcpStream;
266
267		use super::{Stream, StreamOwned};
268		use crate::client::ClientConnection;
269		use crate::server::ServerConnection;
270
271		#[test]
272		fn stream_can_be_created_for_connection_and_tcpstream() {
273		type _Test<'a> = Stream<'a, ClientConnection, TcpStream>;
274		}
275
276		#[test]
277		fn streamowned_can_be_created_for_client_and_tcpstream() {
278		type _Test = StreamOwned<ClientConnection, TcpStream>;
279		}
280
281		#[test]
282		fn streamowned_can_be_created_for_server_and_tcpstream() {
283		type _Test = StreamOwned<ServerConnection, TcpStream>;
284		}
285		}