multipart_write/write/
buffered.rs

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

multipart_write/write/
buffered.rs
