1use core::mem;
2use core::pin::Pin;
3use futures_core::future::{FusedFuture, Future};
4use futures_core::ready;
5use futures_core::stream::{FusedStream, TryStream};
6use futures_core::task::{Context, Poll};
7use pin_project_lite::pin_project;
8
9pin_project! {
10 /// Future for the [`try_collect`](super::TryStreamExt::try_collect) method.
11 #[derive(Debug)]
12 #[must_use = "futures do nothing unless you `.await` or poll them"]
13 pub struct TryCollect<St, C> {
14 #[pin]
15 stream: St,
16 items: C,
17 }
18}
19
20impl<St: TryStream, C: Default> TryCollect<St, C> {
21 pub(super) fn new(s: St) -> Self {
22 Self { stream: s, items: Default::default() }
23 }
24}
25
26impl<St, C> FusedFuture for TryCollect<St, C>
27where
28 St: TryStream + FusedStream,
29 C: Default + Extend<St::Ok>,
30{
31 fn is_terminated(&self) -> bool {
32 self.stream.is_terminated()
33 }
34}
35
36impl<St, C> Future for TryCollect<St, C>
37where
38 St: TryStream,
39 C: Default + Extend<St::Ok>,
40{
41 type Output = Result<C, St::Error>;
42
43 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
44 let mut this: Projection<'_, St, C> = self.project();
45 Poll::Ready(Ok(loop {
46 match ready!(this.stream.as_mut().try_poll_next(cx)?) {
47 Some(x: ::Ok) => this.items.extend(iter:Some(x)),
48 None => break mem::take(dest:this.items),
49 }
50 }))
51 }
52}
53