1 | use super::Feed; |
2 | use core::pin::Pin; |
3 | use futures_core::future::Future; |
4 | use futures_core::ready; |
5 | use futures_core::task::{Context, Poll}; |
6 | use futures_sink::Sink; |
7 | |
8 | /// Future for the [`send`](super::SinkExt::send) method. |
9 | #[derive (Debug)] |
10 | #[must_use = "futures do nothing unless you `.await` or poll them" ] |
11 | pub struct Send<'a, Si: ?Sized, Item> { |
12 | feed: Feed<'a, Si, Item>, |
13 | } |
14 | |
15 | // Pinning is never projected to children |
16 | impl<Si: Unpin + ?Sized, Item> Unpin for Send<'_, Si, Item> {} |
17 | |
18 | impl<'a, Si: Sink<Item> + Unpin + ?Sized, Item> Send<'a, Si, Item> { |
19 | pub(super) fn new(sink: &'a mut Si, item: Item) -> Self { |
20 | Self { feed: Feed::new(sink, item) } |
21 | } |
22 | } |
23 | |
24 | impl<Si: Sink<Item> + Unpin + ?Sized, Item> Future for Send<'_, Si, Item> { |
25 | type Output = Result<(), Si::Error>; |
26 | |
27 | fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |
28 | let this: &mut Send<'_, Si, Item> = &mut *self; |
29 | |
30 | if this.feed.is_item_pending() { |
31 | ready!(Pin::new(&mut this.feed).poll(cx))?; |
32 | debug_assert!(!this.feed.is_item_pending()); |
33 | } |
34 | |
35 | // we're done sending the item, but want to block on flushing the |
36 | // sink |
37 | ready!(this.feed.sink_pin_mut().poll_flush(cx))?; |
38 | |
39 | Poll::Ready(Ok(())) |
40 | } |
41 | } |
42 | |