async_codec_lite/framed/
read.rs

1use super::inner::{FramedInner, ReadFrame};
2use crate::{codec::Decoder, error::Error};
3pub use bytes::{Bytes, BytesMut};
4use futures_core::Stream;
5use futures_io::AsyncRead;
6use futures_sink::Sink;
7use pin_project_lite::pin_project;
8use std::{
9    fmt,
10    pin::Pin,
11    task::{Context, Poll},
12};
13
14pin_project! {
15    pub struct FramedRead<T, D> {
16        #[pin]
17        inner: FramedInner<T, D, ReadFrame>,
18    }
19}
20
21impl<T, D> FramedRead<T, D>
22where
23    T: AsyncRead,
24    D: Decoder,
25{
26    pub fn new(inner: T, decoder: D) -> FramedRead<T, D> {
27        FramedRead {
28            inner: FramedInner {
29                inner,
30                codec: decoder,
31                state: Default::default(),
32            },
33        }
34    }
35
36    pub fn with_capacity(inner: T, decoder: D, capacity: usize) -> FramedRead<T, D> {
37        FramedRead {
38            inner: FramedInner {
39                inner,
40                codec: decoder,
41                state: ReadFrame {
42                    buffer: BytesMut::with_capacity(capacity),
43                    eof: false,
44                    has_errored: false,
45                    is_readable: false,
46                },
47            },
48        }
49    }
50}
51
52impl<T, D> FramedRead<T, D> {
53    pub fn get_ref(&self) -> &T {
54        &self.inner.inner
55    }
56
57    pub fn get_mut(&mut self) -> &mut T {
58        &mut self.inner.inner
59    }
60
61    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut T> {
62        self.project().inner.project().inner
63    }
64
65    pub fn into_inner(self) -> T {
66        self.inner.inner
67    }
68
69    pub fn decoder(&self) -> &D {
70        &self.inner.codec
71    }
72
73    pub fn decoder_mut(&mut self) -> &mut D {
74        &mut self.inner.codec
75    }
76
77    pub fn read_buffer(&self) -> &BytesMut {
78        &self.inner.state.buffer
79    }
80
81    pub fn read_buffer_mut(&mut self) -> &mut BytesMut {
82        &mut self.inner.state.buffer
83    }
84}
85
86impl<T, D> Stream for FramedRead<T, D>
87where
88    T: AsyncRead,
89    D: Decoder,
90{
91    type Item = Result<D::Item, Error<D::Error>>;
92
93    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
94        self.project().inner.poll_next(cx)
95    }
96}
97
98impl<T, I, D> Sink<I> for FramedRead<T, D>
99where
100    T: Sink<I>,
101{
102    type Error = T::Error;
103
104    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
105        self.project().inner.project().inner.poll_ready(cx)
106    }
107
108    fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
109        self.project().inner.project().inner.start_send(item)
110    }
111
112    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
113        self.project().inner.project().inner.poll_flush(cx)
114    }
115
116    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
117        self.project().inner.project().inner.poll_close(cx)
118    }
119}
120
121impl<T, D> fmt::Debug for FramedRead<T, D>
122where
123    T: fmt::Debug,
124    D: fmt::Debug,
125{
126    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
127        f.debug_struct("FramedRead")
128            .field("inner", &self.get_ref())
129            .field("decoder", &self.decoder())
130            .field("buffer", &self.read_buffer())
131            .finish()
132    }
133}
async_codec_lite/framed/read.rs

async_codec_lite/framed/
read.rs
