multipart_write/write/
buffered.rs

1use crate::{FusedMultipartWrite, MultipartWrite};
2
3use futures_core::ready;
4use std::collections::VecDeque;
5use std::fmt::{self, Debug, Formatter};
6use std::pin::Pin;
7use std::task::{Context, Poll};
8
9pin_project_lite::pin_project! {
10    /// `MultipartWrite` for the [`buffered`] method.
11    ///
12    /// [`buffered`]: super::MultipartWriteExt::buffered
13    #[must_use = "futures do nothing unless polled"]
14    pub struct Buffered<Wr, Part> {
15        #[pin]
16        writer: Wr,
17        capacity: usize,
18        buf: VecDeque<Part>,
19    }
20}
21
22impl<Wr, Part> Buffered<Wr, Part> {
23    pub(super) fn new(writer: Wr, capacity: usize) -> Self {
24        Self {
25            writer,
26            capacity,
27            buf: VecDeque::with_capacity(capacity),
28        }
29    }
30
31    /// Consumes `Buffered`, returning the underlying writer.
32    pub fn into_inner(self) -> Wr {
33        self.writer
34    }
35
36    /// Acquires a reference to the underlying writer.
37    pub fn get_ref(&self) -> &Wr {
38        &self.writer
39    }
40
41    /// Acquires a mutable reference to the underlying writer.
42    ///
43    /// It is inadvisable to directly write to the underlying writer.
44    pub fn get_mut(&mut self) -> &mut Wr {
45        &mut self.writer
46    }
47
48    /// Acquires a pinned mutable reference to the underlying writer.
49    ///
50    /// It is inadvisable to directly write to the underlying writer.
51    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut Wr> {
52        self.project().writer
53    }
54
55    fn try_empty_buffer(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Wr::Error>>
56    where
57        Wr: MultipartWrite<Part>,
58    {
59        let mut this = self.project();
60
61        ready!(this.writer.as_mut().poll_ready(cx))?;
62        while let Some(part) = this.buf.pop_front() {
63            this.writer.as_mut().start_send(part)?;
64            if !this.buf.is_empty() {
65                ready!(this.writer.as_mut().poll_ready(cx))?;
66            }
67        }
68        Poll::Ready(Ok(()))
69    }
70}
71
72impl<Wr, Part> FusedMultipartWrite<Part> for Buffered<Wr, Part>
73where
74    Wr: FusedMultipartWrite<Part>,
75{
76    fn is_terminated(&self) -> bool {
77        self.writer.is_terminated()
78    }
79}
80
81impl<Wr, Part> MultipartWrite<Part> for Buffered<Wr, Part>
82where
83    Wr: MultipartWrite<Part>,
84{
85    type Ret = ();
86    type Output = Wr::Output;
87    type Error = Wr::Error;
88
89    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
90        if self.capacity == 0 {
91            return self.project().writer.poll_ready(cx);
92        }
93        let _ = self.as_mut().try_empty_buffer(cx)?;
94        if self.buf.len() >= self.capacity {
95            Poll::Pending
96        } else {
97            Poll::Ready(Ok(()))
98        }
99    }
100
101    fn start_send(self: Pin<&mut Self>, part: Part) -> Result<Self::Ret, Self::Error> {
102        if self.capacity == 0 {
103            let _ = self.project().writer.start_send(part)?;
104        } else {
105            self.project().buf.push_back(part);
106        }
107        Ok(())
108    }
109
110    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
111        ready!(self.as_mut().try_empty_buffer(cx))?;
112        self.project().writer.poll_flush(cx)
113    }
114
115    fn poll_complete(
116        mut self: Pin<&mut Self>,
117        cx: &mut Context<'_>,
118    ) -> Poll<Result<Self::Output, Self::Error>> {
119        ready!(self.as_mut().try_empty_buffer(cx))?;
120        self.project().writer.poll_complete(cx)
121    }
122}
123
124impl<Wr: Debug, Part: Debug> Debug for Buffered<Wr, Part> {
125    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
126        f.debug_struct("Buffered")
127            .field("writer", &self.writer)
128            .field("capacity", &self.capacity)
129            .field("buf", &self.buf)
130            .finish()
131    }
132}
multipart_write/write/buffered.rs

multipart_write/write/
buffered.rs
