futures_util/future/future/
fuse.rs

1use core::pin::Pin;
2use futures_core::future::{FusedFuture, Future};
3use futures_core::task::{Context, Poll};
4use pin_project_lite::pin_project;
5
6pin_project! {
7    /// Future for the [`fuse`](super::FutureExt::fuse) method.
8    #[derive(Debug)]
9    #[must_use = "futures do nothing unless you `.await` or poll them"]
10    pub struct Fuse<Fut> {
11        #[pin]
12        inner: Option<Fut>,
13    }
14}
15
16impl<Fut> Fuse<Fut> {
17    pub(super) fn new(f: Fut) -> Self {
18        Self { inner: Some(f) }
19    }
20}
21
22impl<Fut: Future> Fuse<Fut> {
23    /// Creates a new `Fuse`-wrapped future which is already terminated.
24    ///
25    /// This can be useful in combination with looping and the `select!`
26    /// macro, which bypasses terminated futures.
27    ///
28    /// # Examples
29    ///
30    /// ```
31    /// # futures::executor::block_on(async {
32    /// use core::pin::pin;
33    ///
34    /// use futures::channel::mpsc;
35    /// use futures::future::{Fuse, FusedFuture, FutureExt};
36    /// use futures::select;
37    /// use futures::stream::StreamExt;
38    ///
39    /// let (sender, mut stream) = mpsc::unbounded();
40    ///
41    /// // Send a few messages into the stream
42    /// sender.unbounded_send(()).unwrap();
43    /// sender.unbounded_send(()).unwrap();
44    /// drop(sender);
45    ///
46    /// // Use `Fuse::terminated()` to create an already-terminated future
47    /// // which may be instantiated later.
48    /// let foo_printer = Fuse::terminated();
49    /// let mut foo_printer = pin!(foo_printer);
50    ///
51    /// loop {
52    ///     select! {
53    ///         _ = foo_printer => {},
54    ///         () = stream.select_next_some() => {
55    ///             if !foo_printer.is_terminated() {
56    ///                 println!("Foo is already being printed!");
57    ///             } else {
58    ///                 foo_printer.set(async {
59    ///                     // do some other async operations
60    ///                     println!("Printing foo from `foo_printer` future");
61    ///                 }.fuse());
62    ///             }
63    ///         },
64    ///         complete => break, // `foo_printer` is terminated and the stream is done
65    ///     }
66    /// }
67    /// # });
68    /// ```
69    pub fn terminated() -> Self {
70        Self { inner: None }
71    }
72}
73
74impl<Fut: Future> FusedFuture for Fuse<Fut> {
75    fn is_terminated(&self) -> bool {
76        self.inner.is_none()
77    }
78}
79
80impl<Fut: Future> Future for Fuse<Fut> {
81    type Output = Fut::Output;
82
83    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Fut::Output> {
84        match self.as_mut().project().inner.as_pin_mut() {
85            Some(fut) => fut.poll(cx).map(|output| {
86                self.project().inner.set(None);
87                output
88            }),
89            None => Poll::Pending,
90        }
91    }
92}
futures_util/future/future/fuse.rs

futures_util/future/future/
fuse.rs
