/rust/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-util-0.7.18/src/codec/framed.rs

Source
use crate::codec::decoder::Decoder;
use crate::codec::encoder::Encoder;
use crate::codec::framed_impl::{FramedImpl, RWFrames, ReadFrame, WriteFrame};

use futures_core::Stream;
use tokio::io::{AsyncRead, AsyncWrite};

use bytes::BytesMut;
use futures_sink::Sink;
use pin_project_lite::pin_project;
use std::fmt;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};

pin_project! {
    /// A unified [`Stream`] and [`Sink`] interface to an underlying I/O object, using
    /// the `Encoder` and `Decoder` traits to encode and decode frames.
    ///
    /// You can create a `Framed` instance by using the [`Decoder::framed`] adapter, or
    /// by using the `new` function seen below.
    ///
    /// # Cancellation safety
    ///
    /// * [`futures_util::sink::SinkExt::send`]: if send is used as the event in a
    /// `tokio::select!` statement and some other branch completes first, then it is
    /// guaranteed that the message was not sent, but the message itself is lost.
    /// * [`tokio_stream::StreamExt::next`]: This method is cancel safe. The returned
    /// future only holds onto a reference to the underlying stream, so dropping it will
    /// never lose a value.
    ///
    /// [`Stream`]: futures_core::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`AsyncRead`]: tokio::io::AsyncRead
    /// [`Decoder::framed`]: crate::codec::Decoder::framed()
    /// [`futures_util::sink::SinkExt::send`]: futures_util::sink::SinkExt::send
    /// [`tokio_stream::StreamExt::next`]: https://docs.rs/tokio-stream/latest/tokio_stream/trait.StreamExt.html#method.next
    pub struct Framed<T, U> {
        #[pin]
        inner: FramedImpl<T, U, RWFrames>
    }
}

impl<T, U> Framed<T, U> {
    /// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
    /// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data.
    ///
    /// Raw I/O objects work with byte sequences, but higher-level code usually
    /// wants to batch these into meaningful chunks, called "frames". This
    /// method layers framing on top of an I/O object, by using the codec
    /// traits to handle encoding and decoding of messages frames. Note that
    /// the incoming and outgoing frame types may be distinct.
    ///
    /// This function returns a *single* object that is both [`Stream`] and
    /// [`Sink`]; grouping this into a single object is often useful for layering
    /// things like gzip or TLS, which require both read and write access to the
    /// underlying object.
    ///
    /// If you want to work more directly with the streams and sink, consider
    /// calling [`split`] on the `Framed` returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
    /// Note that, for some byte sources, the stream can be resumed after an EOF
    /// by reading from it, even after it has returned `None`. Repeated attempts
    /// to do so, without new data available, continue to return `None` without
    /// creating more (closing) frames.
    ///
    /// [`Stream`]: futures_core::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Decode`]: crate::codec::Decoder
    /// [`Encoder`]: crate::codec::Encoder
    /// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split
    pub fn new(inner: T, codec: U) -> Framed<T, U> {
        Framed {
            inner: FramedImpl {
                inner,
                codec,
                state: Default::default(),
            },
        }
    }

    /// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
    /// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data,
    /// with a specific read buffer initial capacity.
    ///
    /// Raw I/O objects work with byte sequences, but higher-level code usually
    /// wants to batch these into meaningful chunks, called "frames". This
    /// method layers framing on top of an I/O object, by using the codec
    /// traits to handle encoding and decoding of messages frames. Note that
    /// the incoming and outgoing frame types may be distinct.
    ///
    /// This function returns a *single* object that is both [`Stream`] and
    /// [`Sink`]; grouping this into a single object is often useful for layering
    /// things like gzip or TLS, which require both read and write access to the
    /// underlying object.
    ///
    /// If you want to work more directly with the streams and sink, consider
    /// calling [`split`] on the `Framed` returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
    /// [`Stream`]: futures_core::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Decode`]: crate::codec::Decoder
    /// [`Encoder`]: crate::codec::Encoder
    /// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split
    pub fn with_capacity(inner: T, codec: U, capacity: usize) -> Framed<T, U> {
        Framed {
            inner: FramedImpl {
                inner,
                codec,
                state: RWFrames {
                    read: ReadFrame {
                        eof: false,
                        is_readable: false,
                        buffer: BytesMut::with_capacity(capacity),
                        has_errored: false,
                    },
                    write: WriteFrame {
                        buffer: BytesMut::with_capacity(capacity),
                        backpressure_boundary: capacity,
                    },
                },
            },
        }
    }

    /// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
    /// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data.
    ///
    /// Raw I/O objects work with byte sequences, but higher-level code usually
    /// wants to batch these into meaningful chunks, called "frames". This
    /// method layers framing on top of an I/O object, by using the `Codec`
    /// traits to handle encoding and decoding of messages frames. Note that
    /// the incoming and outgoing frame types may be distinct.
    ///
    /// This function returns a *single* object that is both [`Stream`] and
    /// [`Sink`]; grouping this into a single object is often useful for layering
    /// things like gzip or TLS, which require both read and write access to the
    /// underlying object.
    ///
    /// This objects takes a stream and a `readbuffer` and a `writebuffer`. These field
    /// can be obtained from an existing `Framed` with the [`into_parts`] method.
    ///
    /// If you want to work more directly with the streams and sink, consider
    /// calling [`split`] on the `Framed` returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
    /// [`Stream`]: futures_core::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Decoder`]: crate::codec::Decoder
    /// [`Encoder`]: crate::codec::Encoder
    /// [`into_parts`]: crate::codec::Framed::into_parts()
    /// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split
    pub fn from_parts(parts: FramedParts<T, U>) -> Framed<T, U> {
        Framed {
            inner: FramedImpl {
                inner: parts.io,
                codec: parts.codec,
                state: RWFrames {
                    read: parts.read_buf.into(),
                    write: parts.write_buf.into(),
                },
            },
        }
    }

    /// Returns a reference to the underlying I/O stream wrapped by
    /// `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
    /// of data coming in as it may corrupt the stream of frames otherwise
    /// being worked with.
    pub fn get_ref(&self) -> &T {
        &self.inner.inner
    }

    /// Returns a mutable reference to the underlying I/O stream wrapped by
    /// `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
    /// of data coming in as it may corrupt the stream of frames otherwise
    /// being worked with.
    pub fn get_mut(&mut self) -> &mut T {
        &mut self.inner.inner
    }

    /// Returns a pinned mutable reference to the underlying I/O stream wrapped by
    /// `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
    /// of data coming in as it may corrupt the stream of frames otherwise
    /// being worked with.
    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut T> {
        self.project().inner.project().inner
    }

    /// Returns a reference to the underlying codec wrapped by
    /// `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying codec
    /// as it may corrupt the stream of frames otherwise being worked with.
    pub fn codec(&self) -> &U {
        &self.inner.codec
    }

    /// Returns a mutable reference to the underlying codec wrapped by
    /// `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying codec
    /// as it may corrupt the stream of frames otherwise being worked with.
    pub fn codec_mut(&mut self) -> &mut U {
        &mut self.inner.codec
    }

    /// Maps the codec `U` to `C`, preserving the read and write buffers
    /// wrapped by `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying codec
    /// as it may corrupt the stream of frames otherwise being worked with.
    pub fn map_codec<C, F>(self, map: F) -> Framed<T, C>
    where
        F: FnOnce(U) -> C,
    {
        // This could be potentially simplified once rust-lang/rust#86555 hits stable
        let parts = self.into_parts();
        Framed::from_parts(FramedParts {
            io: parts.io,
            codec: map(parts.codec),
            read_buf: parts.read_buf,
            write_buf: parts.write_buf,
            _priv: (),
        })
    }

    /// Returns a mutable reference to the underlying codec wrapped by
    /// `Framed`.
    ///
    /// Note that care should be taken to not tamper with the underlying codec
    /// as it may corrupt the stream of frames otherwise being worked with.
    pub fn codec_pin_mut(self: Pin<&mut Self>) -> &mut U {
        self.project().inner.project().codec
    }

    /// Returns a reference to the read buffer.
    pub fn read_buffer(&self) -> &BytesMut {
        &self.inner.state.read.buffer
    }

    /// Returns a mutable reference to the read buffer.
    pub fn read_buffer_mut(&mut self) -> &mut BytesMut {
        &mut self.inner.state.read.buffer
    }

    /// Returns a reference to the write buffer.
    pub fn write_buffer(&self) -> &BytesMut {
        &self.inner.state.write.buffer
    }

    /// Returns a mutable reference to the write buffer.
    pub fn write_buffer_mut(&mut self) -> &mut BytesMut {
        &mut self.inner.state.write.buffer
    }

    /// Returns backpressure boundary
    pub fn backpressure_boundary(&self) -> usize {
        self.inner.state.write.backpressure_boundary
    }

    /// Updates backpressure boundary
    pub fn set_backpressure_boundary(&mut self, boundary: usize) {
        self.inner.state.write.backpressure_boundary = boundary;
    }

    /// Consumes the `Framed`, returning its underlying I/O stream.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
    /// of data coming in as it may corrupt the stream of frames otherwise
    /// being worked with.
    pub fn into_inner(self) -> T {
        self.inner.inner
    }

    /// Consumes the `Framed`, returning its underlying I/O stream, the buffer
    /// with unprocessed data, and the codec.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
    /// of data coming in as it may corrupt the stream of frames otherwise
    /// being worked with.
    pub fn into_parts(self) -> FramedParts<T, U> {
        FramedParts {
            io: self.inner.inner,
            codec: self.inner.codec,
            read_buf: self.inner.state.read.buffer,
            write_buf: self.inner.state.write.buffer,
            _priv: (),
        }
    }
}

// This impl just defers to the underlying FramedImpl
impl<T, U> Stream for Framed<T, U>
where
    T: AsyncRead,
    U: Decoder,
{
    type Item = Result<U::Item, U::Error>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.project().inner.poll_next(cx)
    }
}

// This impl just defers to the underlying FramedImpl
impl<T, I, U> Sink<I> for Framed<T, U>
where
    T: AsyncWrite,
    U: Encoder<I>,
    U::Error: From<io::Error>,
{
    type Error = U::Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.project().inner.poll_ready(cx)
    }

    fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
        self.project().inner.start_send(item)
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.project().inner.poll_flush(cx)
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.project().inner.poll_close(cx)
    }
}

impl<T, U> fmt::Debug for Framed<T, U>
where
    T: fmt::Debug,
    U: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Framed")
            .field("io", self.get_ref())
            .field("codec", self.codec())
            .finish()
    }
}

/// `FramedParts` contains an export of the data of a Framed transport.
/// It can be used to construct a new [`Framed`] with a different codec.
/// It contains all current buffers and the inner transport.
///
/// [`Framed`]: crate::codec::Framed
#[derive(Debug)]
#[allow(clippy::manual_non_exhaustive)]
pub struct FramedParts<T, U> {
    /// The inner transport used to read bytes to and write bytes to
    pub io: T,

    /// The codec
    pub codec: U,

    /// The buffer with read but unprocessed data.
    pub read_buf: BytesMut,

    /// A buffer with unprocessed data which are not written yet.
    pub write_buf: BytesMut,

    /// This private field allows us to add additional fields in the future in a
    /// backwards compatible way.
    pub(crate) _priv: (),
}

impl<T, U> FramedParts<T, U> {
    /// Create a new, default, `FramedParts`
    pub fn new<I>(io: T, codec: U) -> FramedParts<T, U>
    where
        U: Encoder<I>,
    {
        FramedParts {
            io,
            codec,
            read_buf: BytesMut::new(),
            write_buf: BytesMut::new(),
            _priv: (),
        }
    }
}

Coverage Report

Created: 2026-02-26 06:09

Line	Count	Source
1		use crate::codec::decoder::Decoder;
2		use crate::codec::encoder::Encoder;
3		use crate::codec::framed_impl::{FramedImpl, RWFrames, ReadFrame, WriteFrame};
4
5		use futures_core::Stream;
6		use tokio::io::{AsyncRead, AsyncWrite};
7
8		use bytes::BytesMut;
9		use futures_sink::Sink;
10		use pin_project_lite::pin_project;
11		use std::fmt;
12		use std::io;
13		use std::pin::Pin;
14		use std::task::{Context, Poll};
15
16		pin_project! {
17		/// A unified [`Stream`] and [`Sink`] interface to an underlying I/O object, using
18		/// the `Encoder` and `Decoder` traits to encode and decode frames.
19		///
20		/// You can create a `Framed` instance by using the [`Decoder::framed`] adapter, or
21		/// by using the `new` function seen below.
22		///
23		/// # Cancellation safety
24		///
25		/// * [`futures_util::sink::SinkExt::send`]: if send is used as the event in a
26		/// `tokio::select!` statement and some other branch completes first, then it is
27		/// guaranteed that the message was not sent, but the message itself is lost.
28		/// * [`tokio_stream::StreamExt::next`]: This method is cancel safe. The returned
29		/// future only holds onto a reference to the underlying stream, so dropping it will
30		/// never lose a value.
31		///
32		/// [`Stream`]: futures_core::Stream
33		/// [`Sink`]: futures_sink::Sink
34		/// [`AsyncRead`]: tokio::io::AsyncRead
35		/// [`Decoder::framed`]: crate::codec::Decoder::framed()
36		/// [`futures_util::sink::SinkExt::send`]: futures_util::sink::SinkExt::send
37		/// [`tokio_stream::StreamExt::next`]: https://docs.rs/tokio-stream/latest/tokio_stream/trait.StreamExt.html#method.next
38		pub struct Framed<T, U> {
39		#[pin]
40		inner: FramedImpl<T, U, RWFrames>
41		}
42		}
43
44		impl<T, U> Framed<T, U> {
45		/// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
46		/// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data.
47		///
48		/// Raw I/O objects work with byte sequences, but higher-level code usually
49		/// wants to batch these into meaningful chunks, called "frames". This
50		/// method layers framing on top of an I/O object, by using the codec
51		/// traits to handle encoding and decoding of messages frames. Note that
52		/// the incoming and outgoing frame types may be distinct.
53		///
54		/// This function returns a single object that is both [`Stream`] and
55		/// [`Sink`]; grouping this into a single object is often useful for layering
56		/// things like gzip or TLS, which require both read and write access to the
57		/// underlying object.
58		///
59		/// If you want to work more directly with the streams and sink, consider
60		/// calling [`split`] on the `Framed` returned by this method, which will
61		/// break them into separate objects, allowing them to interact more easily.
62		///
63		/// Note that, for some byte sources, the stream can be resumed after an EOF
64		/// by reading from it, even after it has returned `None`. Repeated attempts
65		/// to do so, without new data available, continue to return `None` without
66		/// creating more (closing) frames.
67		///
68		/// [`Stream`]: futures_core::Stream
69		/// [`Sink`]: futures_sink::Sink
70		/// [`Decode`]: crate::codec::Decoder
71		/// [`Encoder`]: crate::codec::Encoder
72		/// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split
73	0	pub fn new(inner: T, codec: U) -> Framed<T, U> {
74	0	Framed {
75	0	inner: FramedImpl {
76	0	inner,
77	0	codec,
78	0	state: Default::default(),
79	0	},
80	0	}
81	0	}
82
83		/// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
84		/// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data,
85		/// with a specific read buffer initial capacity.
86		///
87		/// Raw I/O objects work with byte sequences, but higher-level code usually
88		/// wants to batch these into meaningful chunks, called "frames". This
89		/// method layers framing on top of an I/O object, by using the codec
90		/// traits to handle encoding and decoding of messages frames. Note that
91		/// the incoming and outgoing frame types may be distinct.
92		///
93		/// This function returns a single object that is both [`Stream`] and
94		/// [`Sink`]; grouping this into a single object is often useful for layering
95		/// things like gzip or TLS, which require both read and write access to the
96		/// underlying object.
97		///
98		/// If you want to work more directly with the streams and sink, consider
99		/// calling [`split`] on the `Framed` returned by this method, which will
100		/// break them into separate objects, allowing them to interact more easily.
101		///
102		/// [`Stream`]: futures_core::Stream
103		/// [`Sink`]: futures_sink::Sink
104		/// [`Decode`]: crate::codec::Decoder
105		/// [`Encoder`]: crate::codec::Encoder
106		/// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split
107	0	pub fn with_capacity(inner: T, codec: U, capacity: usize) -> Framed<T, U> {
108	0	Framed {
109	0	inner: FramedImpl {
110	0	inner,
111	0	codec,
112	0	state: RWFrames {
113	0	read: ReadFrame {
114	0	eof: false,
115	0	is_readable: false,
116	0	buffer: BytesMut::with_capacity(capacity),
117	0	has_errored: false,
118	0	},
119	0	write: WriteFrame {
120	0	buffer: BytesMut::with_capacity(capacity),
121	0	backpressure_boundary: capacity,
122	0	},
123	0	},
124	0	},
125	0	}
126	0	}
127
128		/// Provides a [`Stream`] and [`Sink`] interface for reading and writing to this
129		/// I/O object, using [`Decoder`] and [`Encoder`] to read and write the raw data.
130		///
131		/// Raw I/O objects work with byte sequences, but higher-level code usually
132		/// wants to batch these into meaningful chunks, called "frames". This
133		/// method layers framing on top of an I/O object, by using the `Codec`
134		/// traits to handle encoding and decoding of messages frames. Note that
135		/// the incoming and outgoing frame types may be distinct.
136		///
137		/// This function returns a single object that is both [`Stream`] and
138		/// [`Sink`]; grouping this into a single object is often useful for layering
139		/// things like gzip or TLS, which require both read and write access to the
140		/// underlying object.
141		///
142		/// This objects takes a stream and a `readbuffer` and a `writebuffer`. These field
143		/// can be obtained from an existing `Framed` with the [`into_parts`] method.
144		///
145		/// If you want to work more directly with the streams and sink, consider
146		/// calling [`split`] on the `Framed` returned by this method, which will
147		/// break them into separate objects, allowing them to interact more easily.
148		///
149		/// [`Stream`]: futures_core::Stream
150		/// [`Sink`]: futures_sink::Sink
151		/// [`Decoder`]: crate::codec::Decoder
152		/// [`Encoder`]: crate::codec::Encoder
153		/// [`into_parts`]: crate::codec::Framed::into_parts()
154		/// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split
155	0	pub fn from_parts(parts: FramedParts<T, U>) -> Framed<T, U> {
156	0	Framed {
157	0	inner: FramedImpl {
158	0	inner: parts.io,
159	0	codec: parts.codec,
160	0	state: RWFrames {
161	0	read: parts.read_buf.into(),
162	0	write: parts.write_buf.into(),
163	0	},
164	0	},
165	0	}
166	0	}
167
168		/// Returns a reference to the underlying I/O stream wrapped by
169		/// `Framed`.
170		///
171		/// Note that care should be taken to not tamper with the underlying stream
172		/// of data coming in as it may corrupt the stream of frames otherwise
173		/// being worked with.
174	0	pub fn get_ref(&self) -> &T {
175	0	&self.inner.inner
176	0	}
177
178		/// Returns a mutable reference to the underlying I/O stream wrapped by
179		/// `Framed`.
180		///
181		/// Note that care should be taken to not tamper with the underlying stream
182		/// of data coming in as it may corrupt the stream of frames otherwise
183		/// being worked with.
184	0	pub fn get_mut(&mut self) -> &mut T {
185	0	&mut self.inner.inner
186	0	}
187
188		/// Returns a pinned mutable reference to the underlying I/O stream wrapped by
189		/// `Framed`.
190		///
191		/// Note that care should be taken to not tamper with the underlying stream
192		/// of data coming in as it may corrupt the stream of frames otherwise
193		/// being worked with.
194	0	pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut T> {
195	0	self.project().inner.project().inner
196	0	}
197
198		/// Returns a reference to the underlying codec wrapped by
199		/// `Framed`.
200		///
201		/// Note that care should be taken to not tamper with the underlying codec
202		/// as it may corrupt the stream of frames otherwise being worked with.
203	0	pub fn codec(&self) -> &U {
204	0	&self.inner.codec
205	0	}
206
207		/// Returns a mutable reference to the underlying codec wrapped by
208		/// `Framed`.
209		///
210		/// Note that care should be taken to not tamper with the underlying codec
211		/// as it may corrupt the stream of frames otherwise being worked with.
212	0	pub fn codec_mut(&mut self) -> &mut U {
213	0	&mut self.inner.codec
214	0	}
215
216		/// Maps the codec `U` to `C`, preserving the read and write buffers
217		/// wrapped by `Framed`.
218		///
219		/// Note that care should be taken to not tamper with the underlying codec
220		/// as it may corrupt the stream of frames otherwise being worked with.
221	0	pub fn map_codec<C, F>(self, map: F) -> Framed<T, C>
222	0	where
223	0	F: FnOnce(U) -> C,
224		{
225		// This could be potentially simplified once rust-lang/rust#86555 hits stable
226	0	let parts = self.into_parts();
227	0	Framed::from_parts(FramedParts {
228	0	io: parts.io,
229	0	codec: map(parts.codec),
230	0	read_buf: parts.read_buf,
231	0	write_buf: parts.write_buf,
232	0	_priv: (),
233	0	})
234	0	}
235
236		/// Returns a mutable reference to the underlying codec wrapped by
237		/// `Framed`.
238		///
239		/// Note that care should be taken to not tamper with the underlying codec
240		/// as it may corrupt the stream of frames otherwise being worked with.
241	0	pub fn codec_pin_mut(self: Pin<&mut Self>) -> &mut U {
242	0	self.project().inner.project().codec
243	0	}
244
245		/// Returns a reference to the read buffer.
246	0	pub fn read_buffer(&self) -> &BytesMut {
247	0	&self.inner.state.read.buffer
248	0	}
249
250		/// Returns a mutable reference to the read buffer.
251	0	pub fn read_buffer_mut(&mut self) -> &mut BytesMut {
252	0	&mut self.inner.state.read.buffer
253	0	}
254
255		/// Returns a reference to the write buffer.
256	0	pub fn write_buffer(&self) -> &BytesMut {
257	0	&self.inner.state.write.buffer
258	0	}
259
260		/// Returns a mutable reference to the write buffer.
261	0	pub fn write_buffer_mut(&mut self) -> &mut BytesMut {
262	0	&mut self.inner.state.write.buffer
263	0	}
264
265		/// Returns backpressure boundary
266	0	pub fn backpressure_boundary(&self) -> usize {
267	0	self.inner.state.write.backpressure_boundary
268	0	}
269
270		/// Updates backpressure boundary
271	0	pub fn set_backpressure_boundary(&mut self, boundary: usize) {
272	0	self.inner.state.write.backpressure_boundary = boundary;
273	0	}
274
275		/// Consumes the `Framed`, returning its underlying I/O stream.
276		///
277		/// Note that care should be taken to not tamper with the underlying stream
278		/// of data coming in as it may corrupt the stream of frames otherwise
279		/// being worked with.
280	0	pub fn into_inner(self) -> T {
281	0	self.inner.inner
282	0	}
283
284		/// Consumes the `Framed`, returning its underlying I/O stream, the buffer
285		/// with unprocessed data, and the codec.
286		///
287		/// Note that care should be taken to not tamper with the underlying stream
288		/// of data coming in as it may corrupt the stream of frames otherwise
289		/// being worked with.
290	0	pub fn into_parts(self) -> FramedParts<T, U> {
291	0	FramedParts {
292	0	io: self.inner.inner,
293	0	codec: self.inner.codec,
294	0	read_buf: self.inner.state.read.buffer,
295	0	write_buf: self.inner.state.write.buffer,
296	0	_priv: (),
297	0	}
298	0	}
299		}
300
301		// This impl just defers to the underlying FramedImpl
302		impl<T, U> Stream for Framed<T, U>
303		where
304		T: AsyncRead,
305		U: Decoder,
306		{
307		type Item = Result<U::Item, U::Error>;
308
309	0	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
310	0	self.project().inner.poll_next(cx)
311	0	}
312		}
313
314		// This impl just defers to the underlying FramedImpl
315		impl<T, I, U> Sink<I> for Framed<T, U>
316		where
317		T: AsyncWrite,
318		U: Encoder<I>,
319		U::Error: From<io::Error>,
320		{
321		type Error = U::Error;
322
323	0	fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
324	0	self.project().inner.poll_ready(cx)
325	0	}
326
327	0	fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
328	0	self.project().inner.start_send(item)
329	0	}
330
331	0	fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
332	0	self.project().inner.poll_flush(cx)
333	0	}
334
335	0	fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
336	0	self.project().inner.poll_close(cx)
337	0	}
338		}
339
340		impl<T, U> fmt::Debug for Framed<T, U>
341		where
342		T: fmt::Debug,
343		U: fmt::Debug,
344		{
345	0	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
346	0	f.debug_struct("Framed")
347	0	.field("io", self.get_ref())
348	0	.field("codec", self.codec())
349	0	.finish()
350	0	}
351		}
352
353		/// `FramedParts` contains an export of the data of a Framed transport.
354		/// It can be used to construct a new [`Framed`] with a different codec.
355		/// It contains all current buffers and the inner transport.
356		///
357		/// [`Framed`]: crate::codec::Framed
358		#[derive(Debug)]
359		#[allow(clippy::manual_non_exhaustive)]
360		pub struct FramedParts<T, U> {
361		/// The inner transport used to read bytes to and write bytes to
362		pub io: T,
363
364		/// The codec
365		pub codec: U,
366
367		/// The buffer with read but unprocessed data.
368		pub read_buf: BytesMut,
369
370		/// A buffer with unprocessed data which are not written yet.
371		pub write_buf: BytesMut,
372
373		/// This private field allows us to add additional fields in the future in a
374		/// backwards compatible way.
375		pub(crate) _priv: (),
376		}
377
378		impl<T, U> FramedParts<T, U> {
379		/// Create a new, default, `FramedParts`
380	0	pub fn new<I>(io: T, codec: U) -> FramedParts<T, U>
381	0	where
382	0	U: Encoder<I>,
383		{
384	0	FramedParts {
385	0	io,
386	0	codec,
387	0	read_buf: BytesMut::new(),
388	0	write_buf: BytesMut::new(),
389	0	_priv: (),
390	0	}
391	0	}
392		}