multipart_write/write/
fuse.rs

1use std::fmt::{self, Debug, Formatter};
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use futures_core::ready;
6
7use crate::{FusedMultipartWrite, MultipartWrite};
8
9pin_project_lite::pin_project! {
10    /// `MultipartWrite` for [`fuse`](super::MultipartWriteExt::fuse).
11    #[must_use = "futures do nothing unless polled"]
12    pub struct Fuse<Wr, F> {
13        #[pin]
14        writer: Wr,
15        f: F,
16        is_terminated: bool,
17    }
18}
19
20impl<Wr, F> Fuse<Wr, F> {
21    pub(super) fn new(writer: Wr, f: F) -> Self {
22        Self { writer, f, is_terminated: false }
23    }
24
25    /// Consumes `Fuse`, returning the underlying writer.
26    pub fn into_inner(self) -> Wr {
27        self.writer
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
50impl<Item, Wr, F> FusedMultipartWrite<Item> for Fuse<Wr, F>
51where
52    Wr: MultipartWrite<Item>,
53    F: FnMut(&Wr::Output) -> bool,
54{
55    fn is_terminated(&self) -> bool {
56        self.is_terminated
57    }
58}
59
60impl<Item, Wr, F> MultipartWrite<Item> for Fuse<Wr, F>
61where
62    Wr: MultipartWrite<Item>,
63    F: FnMut(&Wr::Output) -> bool,
64{
65    type Error = Wr::Error;
66    type Output = Option<Wr::Output>;
67    type Recv = Option<Wr::Recv>;
68
69    fn poll_ready(
70        self: Pin<&mut Self>,
71        cx: &mut Context<'_>,
72    ) -> Poll<Result<(), Self::Error>> {
73        if self.is_terminated {
74            return Poll::Ready(Ok(()));
75        }
76        self.project().writer.as_mut().poll_ready(cx)
77    }
78
79    fn start_send(
80        self: Pin<&mut Self>,
81        part: Item,
82    ) -> Result<Self::Recv, Self::Error> {
83        if self.is_terminated {
84            return Ok(None);
85        }
86        self.project().writer.as_mut().start_send(part).map(Some)
87    }
88
89    fn poll_flush(
90        self: Pin<&mut Self>,
91        cx: &mut Context<'_>,
92    ) -> Poll<Result<(), Self::Error>> {
93        if self.is_terminated {
94            return Poll::Ready(Ok(()));
95        }
96        self.project().writer.as_mut().poll_flush(cx)
97    }
98
99    fn poll_complete(
100        self: Pin<&mut Self>,
101        cx: &mut Context<'_>,
102    ) -> Poll<Result<Self::Output, Self::Error>> {
103        let mut this = self.project();
104        if *this.is_terminated {
105            return Poll::Ready(Ok(None));
106        }
107        let res = ready!(this.writer.as_mut().poll_complete(cx))?;
108        if (this.f)(&res) {
109            *this.is_terminated = true;
110        }
111        Poll::Ready(Ok(Some(res)))
112    }
113}
114
115impl<Wr, F> Debug for Fuse<Wr, F>
116where
117    Wr: Debug,
118{
119    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
120        f.debug_struct("Fuse")
121            .field("writer", &self.writer)
122            .field("f", &"FnMut(&Wr::Output) -> bool")
123            .field("is_terminated", &self.is_terminated)
124            .finish()
125    }
126}
multipart_write/write/fuse.rs

multipart_write/write/
fuse.rs
