From b5b17ed4d800c03beb3ad535d1069a7784e8dc1d Mon Sep 17 00:00:00 2001 From: Héctor Ramón Jiménez Date: Sun, 19 Jan 2020 10:17:08 +0100 Subject: Create `iced_futures` and wire everything up --- futures/Cargo.toml | 23 +++++ futures/src/command.rs | 100 +++++++++++++++++++ futures/src/lib.rs | 8 ++ futures/src/runtime.rs | 79 +++++++++++++++ futures/src/runtime/executor.rs | 26 +++++ futures/src/subscription.rs | 188 ++++++++++++++++++++++++++++++++++++ futures/src/subscription/tracker.rs | 112 +++++++++++++++++++++ 7 files changed, 536 insertions(+) create mode 100644 futures/Cargo.toml create mode 100644 futures/src/command.rs create mode 100644 futures/src/lib.rs create mode 100644 futures/src/runtime.rs create mode 100644 futures/src/runtime/executor.rs create mode 100644 futures/src/subscription.rs create mode 100644 futures/src/subscription/tracker.rs (limited to 'futures') diff --git a/futures/Cargo.toml b/futures/Cargo.toml new file mode 100644 index 00000000..fe0d378c --- /dev/null +++ b/futures/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "iced_futures" +version = "0.1.0-alpha" +authors = ["Héctor Ramón Jiménez "] +edition = "2018" +description = "Commands, subscriptions, and runtimes for Iced" +license = "MIT" +repository = "https://github.com/hecrj/iced" +documentation = "https://docs.rs/iced_futures" +keywords = ["gui", "ui", "graphics", "interface", "futures"] +categories = ["gui"] + +[dependencies] +log = "0.4" + +[dependencies.futures] +version = "0.3" +features = ["thread-pool"] + +[dependencies.tokio] +version = "0.2" +optional = true +features = ["rt-core"] diff --git a/futures/src/command.rs b/futures/src/command.rs new file mode 100644 index 00000000..e7885fb8 --- /dev/null +++ b/futures/src/command.rs @@ -0,0 +1,100 @@ +use futures::future::{BoxFuture, Future, FutureExt}; + +/// A collection of async operations. +/// +/// You should be able to turn a future easily into a [`Command`], either by +/// using the `From` trait or [`Command::perform`]. +/// +/// [`Command`]: struct.Command.html +pub struct Command { + futures: Vec>, +} + +impl Command { + /// Creates an empty [`Command`]. + /// + /// In other words, a [`Command`] that does nothing. + /// + /// [`Command`]: struct.Command.html + pub fn none() -> Self { + Self { + futures: Vec::new(), + } + } + + /// Creates a [`Command`] that performs the action of the given future. + /// + /// [`Command`]: struct.Command.html + pub fn perform( + future: impl Future + 'static + Send, + f: impl Fn(T) -> A + 'static + Send, + ) -> Command { + Command { + futures: vec![future.map(f).boxed()], + } + } + + /// Applies a transformation to the result of a [`Command`]. + /// + /// [`Command`]: struct.Command.html + pub fn map( + mut self, + f: impl Fn(T) -> A + 'static + Send + Sync, + ) -> Command + where + T: 'static, + { + let f = std::sync::Arc::new(f); + + Command { + futures: self + .futures + .drain(..) + .map(|future| { + let f = f.clone(); + + future.map(move |result| f(result)).boxed() + }) + .collect(), + } + } + + /// Creates a [`Command`] that performs the actions of all the given + /// commands. + /// + /// Once this command is run, all the commands will be exectued at once. + /// + /// [`Command`]: struct.Command.html + pub fn batch(commands: impl IntoIterator>) -> Self { + Self { + futures: commands + .into_iter() + .flat_map(|command| command.futures) + .collect(), + } + } + + /// Converts a [`Command`] into its underlying list of futures. + /// + /// [`Command`]: struct.Command.html + pub fn futures(self) -> Vec> { + self.futures + } +} + +impl From for Command +where + A: Future + 'static + Send, +{ + fn from(future: A) -> Self { + Self { + futures: vec![future.boxed()], + } + } +} + +impl std::fmt::Debug for Command { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Command").finish() + } +} diff --git a/futures/src/lib.rs b/futures/src/lib.rs new file mode 100644 index 00000000..f6bcf85a --- /dev/null +++ b/futures/src/lib.rs @@ -0,0 +1,8 @@ +mod command; + +pub mod runtime; +pub mod subscription; + +pub use command::Command; +pub use runtime::Runtime; +pub use subscription::Subscription; diff --git a/futures/src/runtime.rs b/futures/src/runtime.rs new file mode 100644 index 00000000..bc1ad8ac --- /dev/null +++ b/futures/src/runtime.rs @@ -0,0 +1,79 @@ +//! Run commands and subscriptions. +mod executor; + +pub use executor::Executor; + +use crate::{subscription, Command, Subscription}; + +use futures::Sink; +use std::marker::PhantomData; + +#[derive(Debug)] +pub struct Runtime { + executor: Executor, + receiver: Receiver, + subscriptions: subscription::Tracker, + _message: PhantomData, +} + +impl + Runtime +where + Hasher: std::hash::Hasher + Default, + Event: Send + Clone + 'static, + Executor: self::Executor, + Receiver: Sink + + Unpin + + Send + + Clone + + 'static, + Message: Send + 'static, +{ + pub fn new(executor: Executor, receiver: Receiver) -> Self { + Self { + executor, + receiver, + subscriptions: subscription::Tracker::new(), + _message: PhantomData, + } + } + + pub fn enter(&self, f: impl FnOnce() -> R) -> R { + self.executor.enter(f) + } + + pub fn spawn(&mut self, command: Command) { + use futures::{FutureExt, SinkExt}; + + let futures = command.futures(); + + for future in futures { + let mut receiver = self.receiver.clone(); + + self.executor.spawn(future.then(|message| { + async move { + let _ = receiver.send(message).await; + + () + } + })); + } + } + + pub fn track( + &mut self, + subscription: Subscription, + ) { + let futures = self + .subscriptions + .update(subscription, self.receiver.clone()); + + for future in futures { + self.executor.spawn(future); + } + } + + pub fn broadcast(&mut self, event: Event) { + self.subscriptions.broadcast(event); + } +} diff --git a/futures/src/runtime/executor.rs b/futures/src/runtime/executor.rs new file mode 100644 index 00000000..855aa105 --- /dev/null +++ b/futures/src/runtime/executor.rs @@ -0,0 +1,26 @@ +use futures::Future; + +pub trait Executor { + fn spawn(&self, future: impl Future + Send + 'static); + + fn enter(&self, f: impl FnOnce() -> R) -> R { + f() + } +} + +impl Executor for futures::executor::ThreadPool { + fn spawn(&self, future: impl Future + Send + 'static) { + self.spawn_ok(future); + } +} + +#[cfg(feature = "tokio")] +impl Executor for tokio::runtime::Runtime { + fn spawn(&self, future: impl Future + Send + 'static) { + let _ = tokio::runtime::Runtime::spawn(self, future); + } + + fn enter(&self, f: impl FnOnce() -> R) -> R { + tokio::runtime::Runtime::enter(self, f) + } +} diff --git a/futures/src/subscription.rs b/futures/src/subscription.rs new file mode 100644 index 00000000..87e51e48 --- /dev/null +++ b/futures/src/subscription.rs @@ -0,0 +1,188 @@ +//! Listen to external events in your application. +mod tracker; + +pub use tracker::Tracker; + +use futures::stream::BoxStream; + +/// A request to listen to external events. +/// +/// Besides performing async actions on demand with [`Command`], most +/// applications also need to listen to external events passively. +/// +/// A [`Subscription`] is normally provided to some runtime, like a [`Command`], +/// and it will generate events as long as the user keeps requesting it. +/// +/// 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 `Input` and/or +/// `Hasher`. +/// +/// [`Command`]: ../struct.Command.html +/// [`Subscription`]: struct.Subscription.html +pub struct Subscription { + recipes: Vec>>, +} + +impl Subscription +where + H: std::hash::Hasher, +{ + /// Returns an empty [`Subscription`] that will not produce any output. + /// + /// [`Subscription`]: struct.Subscription.html + pub fn none() -> Self { + Self { + recipes: Vec::new(), + } + } + + /// Creates a [`Subscription`] from a [`Recipe`] describing it. + /// + /// [`Subscription`]: struct.Subscription.html + /// [`Recipe`]: trait.Recipe.html + pub fn from_recipe( + recipe: impl Recipe + 'static, + ) -> Self { + Self { + recipes: vec![Box::new(recipe)], + } + } + + /// Batches all the provided subscriptions and returns the resulting + /// [`Subscription`]. + /// + /// [`Subscription`]: struct.Subscription.html + pub fn batch( + subscriptions: impl IntoIterator>, + ) -> Self { + Self { + recipes: subscriptions + .into_iter() + .flat_map(|subscription| subscription.recipes) + .collect(), + } + } + + /// Returns the different recipes of the [`Subscription`]. + /// + /// [`Subscription`]: struct.Subscription.html + pub fn recipes(self) -> Vec>> { + self.recipes + } + + /// Transforms the [`Subscription`] output with the given function. + /// + /// [`Subscription`]: struct.Subscription.html + pub fn map( + mut self, + f: impl Fn(O) -> A + Send + Sync + 'static, + ) -> Subscription + where + H: 'static, + I: 'static, + O: 'static, + A: 'static, + { + let function = std::sync::Arc::new(f); + + Subscription { + recipes: self + .recipes + .drain(..) + .map(|recipe| { + Box::new(Map::new(recipe, function.clone())) + as Box> + }) + .collect(), + } + } +} + +impl std::fmt::Debug for Subscription { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Subscription").finish() + } +} + +/// The description of a [`Subscription`]. +/// +/// A [`Recipe`] is the internal definition of a [`Subscription`]. It is used +/// by runtimes to run and identify subscriptions. You can use it to create your +/// own! +/// +/// [`Subscription`]: struct.Subscription.html +/// [`Recipe`]: trait.Recipe.html +pub trait Recipe { + /// The events that will be produced by a [`Subscription`] with this + /// [`Recipe`]. + /// + /// [`Subscription`]: struct.Subscription.html + /// [`Recipe`]: trait.Recipe.html + type Output; + + /// Hashes the [`Recipe`]. + /// + /// This is used by runtimes to uniquely identify a [`Subscription`]. + /// + /// [`Subscription`]: struct.Subscription.html + /// [`Recipe`]: trait.Recipe.html + fn hash(&self, state: &mut Hasher); + + /// Executes the [`Recipe`] and produces the stream of events of its + /// [`Subscription`]. + /// + /// It receives some generic `Input`, which is normally defined by runtimes. + /// + /// [`Subscription`]: struct.Subscription.html + /// [`Recipe`]: trait.Recipe.html + fn stream( + self: Box, + input: BoxStream<'static, Input>, + ) -> BoxStream<'static, Self::Output>; +} + +struct Map { + recipe: Box>, + mapper: std::sync::Arc B + Send + Sync>, +} + +impl Map { + fn new( + recipe: Box>, + mapper: std::sync::Arc B + Send + Sync + 'static>, + ) -> Self { + Map { recipe, mapper } + } +} + +impl Recipe for Map +where + A: 'static, + B: 'static, + H: std::hash::Hasher, +{ + type Output = B; + + fn hash(&self, state: &mut H) { + use std::hash::Hash; + + std::any::TypeId::of::().hash(state); + self.recipe.hash(state); + } + + fn stream( + self: Box, + input: BoxStream<'static, I>, + ) -> futures::stream::BoxStream<'static, Self::Output> { + use futures::StreamExt; + + let mapper = self.mapper; + + self.recipe + .stream(input) + .map(move |element| mapper(element)) + .boxed() + } +} diff --git a/futures/src/subscription/tracker.rs b/futures/src/subscription/tracker.rs new file mode 100644 index 00000000..a942b619 --- /dev/null +++ b/futures/src/subscription/tracker.rs @@ -0,0 +1,112 @@ +use crate::Subscription; + +use futures::{future::BoxFuture, sink::Sink}; +use std::collections::HashMap; +use std::marker::PhantomData; + +#[derive(Debug)] +pub struct Tracker { + subscriptions: HashMap>, + _hasher: PhantomData, +} + +#[derive(Debug)] +pub struct Execution { + _cancel: futures::channel::oneshot::Sender<()>, + listener: Option>, +} + +impl Tracker +where + Hasher: std::hash::Hasher + Default, + Event: 'static + Send + Clone, +{ + pub fn new() -> Self { + Self { + subscriptions: HashMap::new(), + _hasher: PhantomData, + } + } + + pub fn update( + &mut self, + subscription: Subscription, + receiver: Receiver, + ) -> Vec> + where + Message: 'static + Send, + Receiver: 'static + + Sink + + Unpin + + Send + + Clone, + { + use futures::{future::FutureExt, stream::StreamExt}; + + let mut futures = Vec::new(); + + let recipes = subscription.recipes(); + let mut alive = std::collections::HashSet::new(); + + for recipe in recipes { + let id = { + let mut hasher = Hasher::default(); + recipe.hash(&mut hasher); + + hasher.finish() + }; + + let _ = alive.insert(id); + + if self.subscriptions.contains_key(&id) { + continue; + } + + let (cancel, cancelled) = futures::channel::oneshot::channel(); + + // TODO: Use bus if/when it supports async + let (event_sender, event_receiver) = + futures::channel::mpsc::channel(100); + + let stream = recipe.stream(event_receiver.boxed()); + + let future = futures::future::select( + cancelled, + stream.map(Ok).forward(receiver.clone()), + ) + .map(|_| ()); + + let _ = self.subscriptions.insert( + id, + Execution { + _cancel: cancel, + listener: if event_sender.is_closed() { + None + } else { + Some(event_sender) + }, + }, + ); + + futures.push(future.boxed()); + } + + self.subscriptions.retain(|id, _| alive.contains(&id)); + + futures + } + + pub fn broadcast(&mut self, event: Event) { + self.subscriptions + .values_mut() + .filter_map(|connection| connection.listener.as_mut()) + .for_each(|listener| { + if let Err(error) = listener.try_send(event.clone()) { + log::error!( + "Error sending event to subscription: {:?}", + error + ); + } + }); + } +} -- cgit