Coverage Report

Created: 2024-05-20 06:38

/rust/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.28/src/sink/with.rs
Line
Count
Source (jump to first uncovered line)
1
use core::fmt;
2
use core::marker::PhantomData;
3
use core::pin::Pin;
4
use futures_core::future::Future;
5
use futures_core::ready;
6
use futures_core::stream::Stream;
7
use futures_core::task::{Context, Poll};
8
use futures_sink::Sink;
9
use pin_project_lite::pin_project;
10
11
pin_project! {
12
    /// Sink for the [`with`](super::SinkExt::with) method.
13
    #[must_use = "sinks do nothing unless polled"]
14
    pub struct With<Si, Item, U, Fut, F> {
15
        #[pin]
16
        sink: Si,
17
        f: F,
18
        #[pin]
19
        state: Option<Fut>,
20
        _phantom: PhantomData<fn(U) -> Item>,
21
    }
22
}
23
24
impl<Si, Item, U, Fut, F> fmt::Debug for With<Si, Item, U, Fut, F>
25
where
26
    Si: fmt::Debug,
27
    Fut: fmt::Debug,
28
{
29
0
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30
0
        f.debug_struct("With").field("sink", &self.sink).field("state", &self.state).finish()
31
0
    }
32
}
33
34
impl<Si, Item, U, Fut, F> With<Si, Item, U, Fut, F>
35
where
36
    Si: Sink<Item>,
37
    F: FnMut(U) -> Fut,
38
    Fut: Future,
39
{
40
0
    pub(super) fn new<E>(sink: Si, f: F) -> Self
41
0
    where
42
0
        Fut: Future<Output = Result<Item, E>>,
43
0
        E: From<Si::Error>,
44
0
    {
45
0
        Self { state: None, sink, f, _phantom: PhantomData }
46
0
    }
47
}
48
49
impl<Si, Item, U, Fut, F> Clone for With<Si, Item, U, Fut, F>
50
where
51
    Si: Clone,
52
    F: Clone,
53
    Fut: Clone,
54
{
55
0
    fn clone(&self) -> Self {
56
0
        Self {
57
0
            state: self.state.clone(),
58
0
            sink: self.sink.clone(),
59
0
            f: self.f.clone(),
60
0
            _phantom: PhantomData,
61
0
        }
62
0
    }
63
}
64
65
// Forwarding impl of Stream from the underlying sink
66
impl<S, Item, U, Fut, F> Stream for With<S, Item, U, Fut, F>
67
where
68
    S: Stream + Sink<Item>,
69
    F: FnMut(U) -> Fut,
70
    Fut: Future,
71
{
72
    type Item = S::Item;
73
74
    delegate_stream!(sink);
75
}
76
77
impl<Si, Item, U, Fut, F, E> With<Si, Item, U, Fut, F>
78
where
79
    Si: Sink<Item>,
80
    F: FnMut(U) -> Fut,
81
    Fut: Future<Output = Result<Item, E>>,
82
    E: From<Si::Error>,
83
{
84
    delegate_access_inner!(sink, Si, ());
85
86
    /// Completes the processing of previous item if any.
87
0
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), E>> {
88
0
        let mut this = self.project();
89
90
0
        let item = match this.state.as_mut().as_pin_mut() {
91
0
            None => return Poll::Ready(Ok(())),
92
0
            Some(fut) => ready!(fut.poll(cx))?,
93
        };
94
0
        this.state.set(None);
95
0
        this.sink.start_send(item)?;
96
0
        Poll::Ready(Ok(()))
97
0
    }
98
}
99
100
impl<Si, Item, U, Fut, F, E> Sink<U> for With<Si, Item, U, Fut, F>
101
where
102
    Si: Sink<Item>,
103
    F: FnMut(U) -> Fut,
104
    Fut: Future<Output = Result<Item, E>>,
105
    E: From<Si::Error>,
106
{
107
    type Error = E;
108
109
0
    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
110
0
        ready!(self.as_mut().poll(cx))?;
111
0
        ready!(self.project().sink.poll_ready(cx)?);
112
0
        Poll::Ready(Ok(()))
113
0
    }
114
115
0
    fn start_send(self: Pin<&mut Self>, item: U) -> Result<(), Self::Error> {
116
0
        let mut this = self.project();
117
0
118
0
        assert!(this.state.is_none());
119
0
        this.state.set(Some((this.f)(item)));
120
0
        Ok(())
121
0
    }
122
123
0
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
124
0
        ready!(self.as_mut().poll(cx))?;
125
0
        ready!(self.project().sink.poll_flush(cx)?);
126
0
        Poll::Ready(Ok(()))
127
0
    }
128
129
0
    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
130
0
        ready!(self.as_mut().poll(cx))?;
131
0
        ready!(self.project().sink.poll_close(cx)?);
132
0
        Poll::Ready(Ok(()))
133
0
    }
134
}