summaryrefslogtreecommitdiffstats
path: root/futures/src/subscription.rs
diff options
context:
space:
mode:
Diffstat (limited to 'futures/src/subscription.rs')
-rw-r--r--futures/src/subscription.rs316
1 files changed, 260 insertions, 56 deletions
diff --git a/futures/src/subscription.rs b/futures/src/subscription.rs
index fe53fd7e..876f29c2 100644
--- a/futures/src/subscription.rs
+++ b/futures/src/subscription.rs
@@ -3,7 +3,18 @@ mod tracker;
pub use tracker::Tracker;
-use crate::BoxStream;
+use crate::core::event::{self, Event};
+use crate::core::window;
+use crate::core::Hasher;
+use crate::futures::{Future, Stream};
+use crate::{BoxStream, MaybeSend};
+
+use std::hash::Hash;
+
+/// A stream of runtime events.
+///
+/// It is the input of a [`Subscription`].
+pub type EventStream = BoxStream<(Event, event::Status)>;
/// A request to listen to external events.
///
@@ -16,19 +27,13 @@ use crate::BoxStream;
/// For instance, you can use a [`Subscription`] to listen to a WebSocket
/// connection, keyboard presses, mouse events, time ticks, etc.
///
-/// This type is normally aliased by runtimes with a specific `Event` and/or
-/// `Hasher`.
-///
/// [`Command`]: crate::Command
#[must_use = "`Subscription` must be returned to runtime to take effect"]
-pub struct Subscription<Hasher, Event, Output> {
- recipes: Vec<Box<dyn Recipe<Hasher, Event, Output = Output>>>,
+pub struct Subscription<Message> {
+ recipes: Vec<Box<dyn Recipe<Output = Message>>>,
}
-impl<H, E, O> Subscription<H, E, O>
-where
- H: std::hash::Hasher,
-{
+impl<Message> Subscription<Message> {
/// Returns an empty [`Subscription`] that will not produce any output.
pub fn none() -> Self {
Self {
@@ -38,7 +43,7 @@ where
/// Creates a [`Subscription`] from a [`Recipe`] describing it.
pub fn from_recipe(
- recipe: impl Recipe<H, E, Output = O> + 'static,
+ recipe: impl Recipe<Output = Message> + 'static,
) -> Self {
Self {
recipes: vec![Box::new(recipe)],
@@ -48,7 +53,7 @@ where
/// Batches all the provided subscriptions and returns the resulting
/// [`Subscription`].
pub fn batch(
- subscriptions: impl IntoIterator<Item = Subscription<H, E, O>>,
+ subscriptions: impl IntoIterator<Item = Subscription<Message>>,
) -> Self {
Self {
recipes: subscriptions
@@ -59,18 +64,16 @@ where
}
/// Returns the different recipes of the [`Subscription`].
- pub fn recipes(self) -> Vec<Box<dyn Recipe<H, E, Output = O>>> {
+ pub fn into_recipes(self) -> Vec<Box<dyn Recipe<Output = Message>>> {
self.recipes
}
/// Adds a value to the [`Subscription`] context.
///
/// The value will be part of the identity of a [`Subscription`].
- pub fn with<T>(mut self, value: T) -> Subscription<H, E, (T, O)>
+ pub fn with<T>(mut self, value: T) -> Subscription<(T, Message)>
where
- H: 'static,
- E: 'static,
- O: 'static,
+ Message: 'static,
T: std::hash::Hash + Clone + Send + Sync + 'static,
{
Subscription {
@@ -79,18 +82,16 @@ where
.drain(..)
.map(|recipe| {
Box::new(With::new(recipe, value.clone()))
- as Box<dyn Recipe<H, E, Output = (T, O)>>
+ as Box<dyn Recipe<Output = (T, Message)>>
})
.collect(),
}
}
/// Transforms the [`Subscription`] output with the given function.
- pub fn map<A>(mut self, f: fn(O) -> A) -> Subscription<H, E, A>
+ pub fn map<A>(mut self, f: fn(Message) -> A) -> Subscription<A>
where
- H: 'static,
- E: 'static,
- O: 'static,
+ Message: 'static,
A: 'static,
{
Subscription {
@@ -98,15 +99,14 @@ where
.recipes
.drain(..)
.map(|recipe| {
- Box::new(Map::new(recipe, f))
- as Box<dyn Recipe<H, E, Output = A>>
+ Box::new(Map::new(recipe, f)) as Box<dyn Recipe<Output = A>>
})
.collect(),
}
}
}
-impl<I, O, H> std::fmt::Debug for Subscription<I, O, H> {
+impl<Message> std::fmt::Debug for Subscription<Message> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Subscription").finish()
}
@@ -129,7 +129,7 @@ impl<I, O, H> std::fmt::Debug for Subscription<I, O, H> {
/// [examples]: https://github.com/iced-rs/iced/tree/0.8/examples
/// [`download_progress`]: https://github.com/iced-rs/iced/tree/0.8/examples/download_progress
/// [`stopwatch`]: https://github.com/iced-rs/iced/tree/0.8/examples/stopwatch
-pub trait Recipe<Hasher: std::hash::Hasher, Event> {
+pub trait Recipe {
/// The events that will be produced by a [`Subscription`] with this
/// [`Recipe`].
type Output;
@@ -141,45 +141,33 @@ pub trait Recipe<Hasher: std::hash::Hasher, Event> {
/// Executes the [`Recipe`] and produces the stream of events of its
/// [`Subscription`].
- ///
- /// It receives some stream of generic events, which is normally defined by
- /// shells.
- fn stream(
- self: Box<Self>,
- input: BoxStream<Event>,
- ) -> BoxStream<Self::Output>;
+ fn stream(self: Box<Self>, input: EventStream) -> BoxStream<Self::Output>;
}
-struct Map<Hasher, Event, A, B> {
- recipe: Box<dyn Recipe<Hasher, Event, Output = A>>,
+struct Map<A, B> {
+ recipe: Box<dyn Recipe<Output = A>>,
mapper: fn(A) -> B,
}
-impl<H, E, A, B> Map<H, E, A, B> {
- fn new(
- recipe: Box<dyn Recipe<H, E, Output = A>>,
- mapper: fn(A) -> B,
- ) -> Self {
+impl<A, B> Map<A, B> {
+ fn new(recipe: Box<dyn Recipe<Output = A>>, mapper: fn(A) -> B) -> Self {
Map { recipe, mapper }
}
}
-impl<H, E, A, B> Recipe<H, E> for Map<H, E, A, B>
+impl<A, B> Recipe for Map<A, B>
where
A: 'static,
B: 'static,
- H: std::hash::Hasher,
{
type Output = B;
- fn hash(&self, state: &mut H) {
- use std::hash::Hash;
-
+ fn hash(&self, state: &mut Hasher) {
self.recipe.hash(state);
self.mapper.hash(state);
}
- fn stream(self: Box<Self>, input: BoxStream<E>) -> BoxStream<Self::Output> {
+ fn stream(self: Box<Self>, input: EventStream) -> BoxStream<Self::Output> {
use futures::StreamExt;
let mapper = self.mapper;
@@ -188,34 +176,31 @@ where
}
}
-struct With<Hasher, Event, A, B> {
- recipe: Box<dyn Recipe<Hasher, Event, Output = A>>,
+struct With<A, B> {
+ recipe: Box<dyn Recipe<Output = A>>,
value: B,
}
-impl<H, E, A, B> With<H, E, A, B> {
- fn new(recipe: Box<dyn Recipe<H, E, Output = A>>, value: B) -> Self {
+impl<A, B> With<A, B> {
+ fn new(recipe: Box<dyn Recipe<Output = A>>, value: B) -> Self {
With { recipe, value }
}
}
-impl<H, E, A, B> Recipe<H, E> for With<H, E, A, B>
+impl<A, B> Recipe for With<A, B>
where
A: 'static,
B: 'static + std::hash::Hash + Clone + Send + Sync,
- H: std::hash::Hasher,
{
type Output = (B, A);
- fn hash(&self, state: &mut H) {
- use std::hash::Hash;
-
+ fn hash(&self, state: &mut Hasher) {
std::any::TypeId::of::<B>().hash(state);
self.value.hash(state);
self.recipe.hash(state);
}
- fn stream(self: Box<Self>, input: BoxStream<E>) -> BoxStream<Self::Output> {
+ fn stream(self: Box<Self>, input: EventStream) -> BoxStream<Self::Output> {
use futures::StreamExt;
let value = self.value;
@@ -227,3 +212,222 @@ where
)
}
}
+
+/// Returns a [`Subscription`] to all the ignored runtime events.
+///
+/// This subscription will notify your application of any [`Event`] that was
+/// not captured by any widget.
+pub fn events() -> Subscription<Event> {
+ events_with(|event, status| match status {
+ event::Status::Ignored => Some(event),
+ event::Status::Captured => None,
+ })
+}
+
+/// Returns a [`Subscription`] that filters all the runtime events with the
+/// provided function, producing messages accordingly.
+///
+/// This subscription will call the provided function for every [`Event`]
+/// handled by the runtime. If the function:
+///
+/// - Returns `None`, the [`Event`] will be discarded.
+/// - Returns `Some` message, the `Message` will be produced.
+pub fn events_with<Message>(
+ f: fn(Event, event::Status) -> Option<Message>,
+) -> Subscription<Message>
+where
+ Message: 'static + MaybeSend,
+{
+ #[derive(Hash)]
+ struct EventsWith;
+
+ Subscription::from_recipe(Runner {
+ id: (EventsWith, f),
+ spawn: move |events| {
+ use futures::future;
+ use futures::stream::StreamExt;
+
+ events.filter_map(move |(event, status)| {
+ future::ready(match event {
+ Event::Window(window::Event::RedrawRequested(_)) => None,
+ _ => f(event, status),
+ })
+ })
+ },
+ })
+}
+
+/// Returns a [`Subscription`] that produces a message for every runtime event,
+/// including the redraw request events.
+///
+/// **Warning:** This [`Subscription`], if unfiltered, may produce messages in
+/// an infinite loop.
+pub fn raw_events<Message>(
+ f: fn(Event, event::Status) -> Option<Message>,
+) -> Subscription<Message>
+where
+ Message: 'static + MaybeSend,
+{
+ #[derive(Hash)]
+ struct RawEvents;
+
+ Subscription::from_recipe(Runner {
+ id: (RawEvents, f),
+ spawn: move |events| {
+ use futures::future;
+ use futures::stream::StreamExt;
+
+ events.filter_map(move |(event, status)| {
+ future::ready(f(event, status))
+ })
+ },
+ })
+}
+
+/// Returns a [`Subscription`] that will call the given function to create and
+/// asynchronously run the given [`Stream`].
+pub fn run<S, Message>(builder: fn() -> S) -> Subscription<Message>
+where
+ S: Stream<Item = Message> + MaybeSend + 'static,
+ Message: 'static,
+{
+ Subscription::from_recipe(Runner {
+ id: builder,
+ spawn: move |_| builder(),
+ })
+}
+
+/// Returns a [`Subscription`] that will create and asynchronously run the
+/// given [`Stream`].
+///
+/// The `id` will be used to uniquely identify the [`Subscription`].
+pub fn run_with_id<I, S, Message>(id: I, stream: S) -> Subscription<Message>
+where
+ I: Hash + 'static,
+ S: Stream<Item = Message> + MaybeSend + 'static,
+ Message: 'static,
+{
+ Subscription::from_recipe(Runner {
+ id,
+ spawn: move |_| stream,
+ })
+}
+
+/// Returns a [`Subscription`] that will create and asynchronously run a
+/// [`Stream`] that will call the provided closure to produce every `Message`.
+///
+/// The `id` will be used to uniquely identify the [`Subscription`].
+///
+/// # Creating an asynchronous worker with bidirectional communication
+/// You can leverage this helper to create a [`Subscription`] that spawns
+/// an asynchronous worker in the background and establish a channel of
+/// communication with an `iced` application.
+///
+/// You can achieve this by creating an `mpsc` channel inside the closure
+/// and returning the `Sender` as a `Message` for the `Application`:
+///
+/// ```
+/// use iced_futures::subscription::{self, Subscription};
+/// use iced_futures::futures;
+///
+/// use futures::channel::mpsc;
+///
+/// pub enum Event {
+/// Ready(mpsc::Sender<Input>),
+/// WorkFinished,
+/// // ...
+/// }
+///
+/// enum Input {
+/// DoSomeWork,
+/// // ...
+/// }
+///
+/// enum State {
+/// Starting,
+/// Ready(mpsc::Receiver<Input>),
+/// }
+///
+/// fn some_worker() -> Subscription<Event> {
+/// struct SomeWorker;
+///
+/// subscription::unfold(std::any::TypeId::of::<SomeWorker>(), State::Starting, |state| async move {
+/// match state {
+/// State::Starting => {
+/// // Create channel
+/// let (sender, receiver) = mpsc::channel(100);
+///
+/// (Some(Event::Ready(sender)), State::Ready(receiver))
+/// }
+/// State::Ready(mut receiver) => {
+/// use futures::StreamExt;
+///
+/// // Read next input sent from `Application`
+/// let input = receiver.select_next_some().await;
+///
+/// match input {
+/// Input::DoSomeWork => {
+/// // Do some async work...
+///
+/// // Finally, we can optionally return a message to tell the
+/// // `Application` the work is done
+/// (Some(Event::WorkFinished), State::Ready(receiver))
+/// }
+/// }
+/// }
+/// }
+/// })
+/// }
+/// ```
+///
+/// Check out the [`websocket`] example, which showcases this pattern to maintain a WebSocket
+/// connection open.
+///
+/// [`websocket`]: https://github.com/iced-rs/iced/tree/0.8/examples/websocket
+pub fn unfold<I, T, Fut, Message>(
+ id: I,
+ initial: T,
+ mut f: impl FnMut(T) -> Fut + MaybeSend + Sync + 'static,
+) -> Subscription<Message>
+where
+ I: Hash + 'static,
+ T: MaybeSend + 'static,
+ Fut: Future<Output = (Option<Message>, T)> + MaybeSend + 'static,
+ Message: 'static + MaybeSend,
+{
+ use futures::future::{self, FutureExt};
+ use futures::stream::StreamExt;
+
+ run_with_id(
+ id,
+ futures::stream::unfold(initial, move |state| f(state).map(Some))
+ .filter_map(future::ready),
+ )
+}
+
+struct Runner<I, F, S, Message>
+where
+ F: FnOnce(EventStream) -> S,
+ S: Stream<Item = Message>,
+{
+ id: I,
+ spawn: F,
+}
+
+impl<I, S, F, Message> Recipe for Runner<I, F, S, Message>
+where
+ I: Hash + 'static,
+ F: FnOnce(EventStream) -> S,
+ S: Stream<Item = Message> + MaybeSend + 'static,
+{
+ type Output = Message;
+
+ fn hash(&self, state: &mut Hasher) {
+ std::any::TypeId::of::<I>().hash(state);
+ self.id.hash(state);
+ }
+
+ fn stream(self: Box<Self>, input: EventStream) -> BoxStream<Self::Output> {
+ crate::boxed_stream((self.spawn)(input))
+ }
+}
generated by cgit (git 2.34.1) at 2025-04-09 03:44:56 +0000