1use crate::codec::framed_impl::{FramedImpl, ReadFrame};
2use crate::codec::Decoder;
3
4use futures_core::Stream;
5use tokio::io::AsyncRead;
6
7use bytes::BytesMut;
8use futures_sink::Sink;
9use pin_project_lite::pin_project;
10use std::fmt;
11use std::pin::Pin;
12use std::task::{Context, Poll};
13
14pin_project! {
15 /// A [`Stream`] of messages decoded from an [`AsyncRead`].
16 ///
17 /// For examples of how to use `FramedRead` with a codec, see the
18 /// examples on the [`codec`] module.
19 ///
20 /// # Cancellation safety
21 /// * [`tokio_stream::StreamExt::next`]: This method is cancel safe. The returned
22 /// future only holds onto a reference to the underlying stream, so dropping it will
23 /// never lose a value.
24 ///
25 /// [`Stream`]: futures_core::Stream
26 /// [`AsyncRead`]: tokio::io::AsyncRead
27 /// [`codec`]: crate::codec
28 /// [`tokio_stream::StreamExt::next`]: https://docs.rs/tokio-stream/latest/tokio_stream/trait.StreamExt.html#method.next
29 pub struct FramedRead<T, D> {
30 #[pin]
31 inner: FramedImpl<T, D, ReadFrame>,
32 }
33}
34
35// ===== impl FramedRead =====
36
37impl<T, D> FramedRead<T, D>
38where
39 T: AsyncRead,
40 D: Decoder,
41{
42 /// Creates a new `FramedRead` with the given `decoder`.
43 pub fn new(inner: T, decoder: D) -> FramedRead<T, D> {
44 FramedRead {
45 inner: FramedImpl {
46 inner,
47 codec: decoder,
48 state: Default::default(),
49 },
50 }
51 }
52
53 /// Creates a new `FramedRead` with the given `decoder` and a buffer of `capacity`
54 /// initial size.
55 pub fn with_capacity(inner: T, decoder: D, capacity: usize) -> FramedRead<T, D> {
56 FramedRead {
57 inner: FramedImpl {
58 inner,
59 codec: decoder,
60 state: ReadFrame {
61 eof: false,
62 is_readable: false,
63 buffer: BytesMut::with_capacity(capacity),
64 has_errored: false,
65 },
66 },
67 }
68 }
69}
70
71impl<T, D> FramedRead<T, D> {
72 /// Returns a reference to the underlying I/O stream wrapped by
73 /// `FramedRead`.
74 ///
75 /// Note that care should be taken to not tamper with the underlying stream
76 /// of data coming in as it may corrupt the stream of frames otherwise
77 /// being worked with.
78 pub fn get_ref(&self) -> &T {
79 &self.inner.inner
80 }
81
82 /// Returns a mutable reference to the underlying I/O stream wrapped by
83 /// `FramedRead`.
84 ///
85 /// Note that care should be taken to not tamper with the underlying stream
86 /// of data coming in as it may corrupt the stream of frames otherwise
87 /// being worked with.
88 pub fn get_mut(&mut self) -> &mut T {
89 &mut self.inner.inner
90 }
91
92 /// Returns a pinned mutable reference to the underlying I/O stream wrapped by
93 /// `FramedRead`.
94 ///
95 /// Note that care should be taken to not tamper with the underlying stream
96 /// of data coming in as it may corrupt the stream of frames otherwise
97 /// being worked with.
98 pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut T> {
99 self.project().inner.project().inner
100 }
101
102 /// Consumes the `FramedRead`, returning its underlying I/O stream.
103 ///
104 /// Note that care should be taken to not tamper with the underlying stream
105 /// of data coming in as it may corrupt the stream of frames otherwise
106 /// being worked with.
107 pub fn into_inner(self) -> T {
108 self.inner.inner
109 }
110
111 /// Returns a reference to the underlying decoder.
112 pub fn decoder(&self) -> &D {
113 &self.inner.codec
114 }
115
116 /// Returns a mutable reference to the underlying decoder.
117 pub fn decoder_mut(&mut self) -> &mut D {
118 &mut self.inner.codec
119 }
120
121 /// Maps the decoder `D` to `C`, preserving the read buffer
122 /// wrapped by `Framed`.
123 pub fn map_decoder<C, F>(self, map: F) -> FramedRead<T, C>
124 where
125 F: FnOnce(D) -> C,
126 {
127 // This could be potentially simplified once rust-lang/rust#86555 hits stable
128 let FramedImpl {
129 inner,
130 state,
131 codec,
132 } = self.inner;
133 FramedRead {
134 inner: FramedImpl {
135 inner,
136 state,
137 codec: map(codec),
138 },
139 }
140 }
141
142 /// Returns a mutable reference to the underlying decoder.
143 pub fn decoder_pin_mut(self: Pin<&mut Self>) -> &mut D {
144 self.project().inner.project().codec
145 }
146
147 /// Returns a reference to the read buffer.
148 pub fn read_buffer(&self) -> &BytesMut {
149 &self.inner.state.buffer
150 }
151
152 /// Returns a mutable reference to the read buffer.
153 pub fn read_buffer_mut(&mut self) -> &mut BytesMut {
154 &mut self.inner.state.buffer
155 }
156}
157
158// This impl just defers to the underlying FramedImpl
159impl<T, D> Stream for FramedRead<T, D>
160where
161 T: AsyncRead,
162 D: Decoder,
163{
164 type Item = Result<D::Item, D::Error>;
165
166 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
167 self.project().inner.poll_next(cx)
168 }
169}
170
171// This impl just defers to the underlying T: Sink
172impl<T, I, D> Sink<I> for FramedRead<T, D>
173where
174 T: Sink<I>,
175{
176 type Error = T::Error;
177
178 fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
179 self.project().inner.project().inner.poll_ready(cx)
180 }
181
182 fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
183 self.project().inner.project().inner.start_send(item)
184 }
185
186 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
187 self.project().inner.project().inner.poll_flush(cx)
188 }
189
190 fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
191 self.project().inner.project().inner.poll_close(cx)
192 }
193}
194
195impl<T, D> fmt::Debug for FramedRead<T, D>
196where
197 T: fmt::Debug,
198 D: fmt::Debug,
199{
200 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
201 f&mut DebugStruct<'_, '_>.debug_struct("FramedRead")
202 .field("inner", &self.get_ref())
203 .field("decoder", &self.decoder())
204 .field("eof", &self.inner.state.eof)
205 .field("is_readable", &self.inner.state.is_readable)
206 .field(name:"buffer", &self.read_buffer())
207 .finish()
208 }
209}
210