iced_futures/
runtime.rs

Help
1//! Run commands and keep track of subscriptions.
2use crate::subscription;
3use crate::{BoxStream, Executor, MaybeSend};
4
5use futures::{Sink, channel::mpsc};
6use std::marker::PhantomData;
7
8/// A batteries-included runtime of commands and subscriptions.
9///
10/// If you have an [`Executor`], a [`Runtime`] can be leveraged to run any
11/// `Command` or [`Subscription`] and get notified of the results!
12///
13/// [`Subscription`]: crate::Subscription
14#[derive(Debug)]
15pub struct Runtime<Executor, Sender, Message> {
16    executor: Executor,
17    sender: Sender,
18    subscriptions: subscription::Tracker,
19    _message: PhantomData<Message>,
20}
21
22impl<Executor, Sender, Message> Runtime<Executor, Sender, Message>
23where
24    Executor: self::Executor,
25    Sender: Sink<Message, Error = mpsc::SendError>
26        + Unpin
27        + MaybeSend
28        + Clone
29        + 'static,
30    Message: MaybeSend + 'static,
31{
32    /// Creates a new empty [`Runtime`].
33    ///
34    /// You need to provide:
35    /// - an [`Executor`] to spawn futures
36    /// - a `Sender` implementing `Sink` to receive the results
37    pub fn new(executor: Executor, sender: Sender) -> Self {
38        Self {
39            executor,
40            sender,
41            subscriptions: subscription::Tracker::new(),
42            _message: PhantomData,
43        }
44    }
45
46    /// Runs the given closure inside the [`Executor`] of the [`Runtime`].
47    ///
48    /// See [`Executor::enter`] to learn more.
49    pub fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
50        self.executor.enter(f)
51    }
52
53    /// Runs a future to completion in the current thread within the [`Runtime`].
54    #[cfg(not(target_arch = "wasm32"))]
55    pub fn block_on<T>(&mut self, future: impl Future<Output = T>) -> T {
56        self.executor.block_on(future)
57    }
58
59    /// Runs a [`Stream`] in the [`Runtime`] until completion.
60    ///
61    /// The resulting `Message`s will be forwarded to the `Sender` of the
62    /// [`Runtime`].
63    ///
64    /// [`Stream`]: BoxStream
65    pub fn run(&mut self, stream: BoxStream<Message>) {
66        use futures::{FutureExt, StreamExt};
67
68        let sender = self.sender.clone();
69        let future =
70            stream.map(Ok).forward(sender).map(|result| match result {
71                Ok(()) => (),
72                Err(error) => {
73                    log::warn!(
74                        "Stream could not run until completion: {error}"
75                    );
76                }
77            });
78
79        self.executor.spawn(future);
80    }
81
82    /// Tracks a [`Subscription`] in the [`Runtime`].
83    ///
84    /// It will spawn new streams or close old ones as necessary! See
85    /// [`Tracker::update`] to learn more about this!
86    ///
87    /// [`Tracker::update`]: subscription::Tracker::update
88    /// [`Subscription`]: crate::Subscription
89    pub fn track(
90        &mut self,
91        recipes: impl IntoIterator<
92            Item = Box<dyn subscription::Recipe<Output = Message>>,
93        >,
94    ) {
95        let Runtime {
96            executor,
97            subscriptions,
98            sender,
99            ..
100        } = self;
101
102        let futures = executor.enter(|| {
103            subscriptions.update(recipes.into_iter(), sender.clone())
104        });
105
106        for future in futures {
107            executor.spawn(future);
108        }
109    }
110
111    /// Broadcasts an event to all the subscriptions currently alive in the
112    /// [`Runtime`].
113    ///
114    /// See [`Tracker::broadcast`] to learn more.
115    ///
116    /// [`Tracker::broadcast`]: subscription::Tracker::broadcast
117    pub fn broadcast(&mut self, event: subscription::Event) {
118        self.subscriptions.broadcast(event);
119    }
120}
iced_futures/runtime.rs

iced_futures/
runtime.rs
