completion/stream/futures/
next.rs

1use core::future::Future;
2use core::pin::Pin;
3use core::task::{Context, Poll};
4
5use completion_core::{CompletionFuture, CompletionStream};
6use futures_core::Stream;
7
8use super::super::CompletionStreamExt;
9
10/// Future for [`CompletionStreamExt::next`].
11#[derive(Debug)]
12pub struct Next<'a, S: ?Sized> {
13    stream: &'a mut S,
14}
15
16impl<'a, S: ?Sized> Next<'a, S> {
17    pub(crate) fn new(stream: &'a mut S) -> Self {
18        Self { stream }
19    }
20}
21
22impl<S: Unpin + ?Sized> Unpin for Next<'_, S> {}
23
24impl<S: Unpin + ?Sized> CompletionFuture for Next<'_, S>
25where
26    S: CompletionStream,
27{
28    type Output = Option<S::Item>;
29
30    unsafe fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
31        self.stream.poll_next(cx)
32    }
33    unsafe fn poll_cancel(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
34        self.stream.poll_cancel(cx)
35    }
36}
37
38impl<S: Unpin + ?Sized> Future for Next<'_, S>
39where
40    S: CompletionStream + Stream<Item = <S as CompletionStream>::Item>,
41{
42    type Output = Option<<S as CompletionStream>::Item>;
43
44    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
45        unsafe { CompletionFuture::poll(self, cx) }
46    }
47}
completion/stream/futures/next.rs

completion/stream/futures/
next.rs
