From 3b7d479534d9114ed12bb5d9ccd910e85d5c13c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9ctor=20Ram=C3=B3n=20Jim=C3=A9nez?= Date: Wed, 29 Nov 2023 00:12:48 +0100 Subject: [PATCH 1/2] Implement `Command::run` for executing a `Stream` to completion --- futures/src/runtime.rs | 25 ++++++++++++++++++++++++- runtime/src/command.rs | 14 +++++++++++++- runtime/src/command/action.rs | 9 ++++++++- winit/src/application.rs | 3 +++ 4 files changed, 48 insertions(+), 3 deletions(-) diff --git a/futures/src/runtime.rs b/futures/src/runtime.rs index 16111b3650..cac7b7e19e 100644 --- a/futures/src/runtime.rs +++ b/futures/src/runtime.rs @@ -1,7 +1,7 @@ //! Run commands and keep track of subscriptions. use crate::core::event::{self, Event}; use crate::subscription; -use crate::{BoxFuture, Executor, MaybeSend}; +use crate::{BoxFuture, BoxStream, Executor, MaybeSend}; use futures::{channel::mpsc, Sink}; use std::marker::PhantomData; @@ -69,6 +69,29 @@ where self.executor.spawn(future); } + /// Runs a [`Stream`] in the [`Runtime`] until completion. + /// + /// The resulting `Message`s will be forwarded to the `Sender` of the + /// [`Runtime`]. + /// + /// [`Stream`]: BoxStream + pub fn run(&mut self, stream: BoxStream) { + use futures::{FutureExt, StreamExt}; + + let sender = self.sender.clone(); + let future = + stream.map(Ok).forward(sender).map(|result| match result { + Ok(()) => (), + Err(error) => { + log::warn!( + "Stream could not run until completion: {error}" + ); + } + }); + + self.executor.spawn(future); + } + /// Tracks a [`Subscription`] in the [`Runtime`]. /// /// It will spawn new streams or close old ones as necessary! See diff --git a/runtime/src/command.rs b/runtime/src/command.rs index b74097bdfc..b942f3ced6 100644 --- a/runtime/src/command.rs +++ b/runtime/src/command.rs @@ -4,8 +4,10 @@ mod action; pub use action::Action; use crate::core::widget; +use crate::futures::futures; use crate::futures::MaybeSend; +use futures::Stream; use std::fmt; use std::future::Future; @@ -43,11 +45,21 @@ impl Command { future: impl Future + 'static + MaybeSend, f: impl FnOnce(A) -> T + 'static + MaybeSend, ) -> Command { - use iced_futures::futures::FutureExt; + use futures::FutureExt; Command::single(Action::Future(Box::pin(future.map(f)))) } + /// Creates a [`Command`] that runs the given stream to completion. + pub fn run( + stream: impl Stream + 'static + MaybeSend, + f: impl Fn(A) -> T + 'static + MaybeSend, + ) -> Command { + use futures::StreamExt; + + Command::single(Action::Stream(Box::pin(stream.map(f)))) + } + /// Creates a [`Command`] that performs the actions of all the given /// commands. /// diff --git a/runtime/src/command/action.rs b/runtime/src/command/action.rs index 6c74f0efb2..6551e23302 100644 --- a/runtime/src/command/action.rs +++ b/runtime/src/command/action.rs @@ -18,6 +18,11 @@ pub enum Action { /// [`Future`]: iced_futures::BoxFuture Future(iced_futures::BoxFuture), + /// Run a [`Stream`] to completion. + /// + /// [`Stream`]: iced_futures::BoxStream + Stream(iced_futures::BoxStream), + /// Run a clipboard action. Clipboard(clipboard::Action), @@ -52,10 +57,11 @@ impl Action { A: 'static, T: 'static, { - use iced_futures::futures::FutureExt; + use iced_futures::futures::{FutureExt, StreamExt}; match self { Self::Future(future) => Action::Future(Box::pin(future.map(f))), + Self::Stream(stream) => Action::Stream(Box::pin(stream.map(f))), Self::Clipboard(action) => Action::Clipboard(action.map(f)), Self::Window(window) => Action::Window(window.map(f)), Self::System(system) => Action::System(system.map(f)), @@ -74,6 +80,7 @@ impl fmt::Debug for Action { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Future(_) => write!(f, "Action::Future"), + Self::Stream(_) => write!(f, "Action::Stream"), Self::Clipboard(action) => { write!(f, "Action::Clipboard({action:?})") } diff --git a/winit/src/application.rs b/winit/src/application.rs index 315e34d93e..2c5c864ac0 100644 --- a/winit/src/application.rs +++ b/winit/src/application.rs @@ -736,6 +736,9 @@ pub fn run_command( command::Action::Future(future) => { runtime.spawn(future); } + command::Action::Stream(stream) => { + runtime.run(stream); + } command::Action::Clipboard(action) => match action { clipboard::Action::Read(tag) => { let message = tag(clipboard.read()); From a761448858521d11dc646e2ef5217e9e06628932 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9ctor=20Ram=C3=B3n=20Jim=C3=A9nez?= Date: Wed, 29 Nov 2023 00:14:27 +0100 Subject: [PATCH 2/2] Implement `command::channel` helper It is analogous to `subscription::channel`. --- runtime/src/command.rs | 21 +++++++++++++++++++++ src/lib.rs | 7 ++++++- 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/runtime/src/command.rs b/runtime/src/command.rs index b942f3ced6..f70da915fb 100644 --- a/runtime/src/command.rs +++ b/runtime/src/command.rs @@ -7,6 +7,7 @@ use crate::core::widget; use crate::futures::futures; use crate::futures::MaybeSend; +use futures::channel::mpsc; use futures::Stream; use std::fmt; use std::future::Future; @@ -118,3 +119,23 @@ impl fmt::Debug for Command { command.fmt(f) } } + +/// Creates a [`Command`] that produces the `Message`s published from a [`Future`] +/// to an [`mpsc::Sender`] with the given bounds. +pub fn channel( + size: usize, + f: impl FnOnce(mpsc::Sender) -> Fut + MaybeSend + 'static, +) -> Command +where + Fut: Future + MaybeSend + 'static, + Message: 'static + MaybeSend, +{ + use futures::future; + use futures::stream::{self, StreamExt}; + + let (sender, receiver) = mpsc::channel(size); + + let runner = stream::once(f(sender)).filter_map(|_| future::ready(None)); + + Command::single(Action::Stream(Box::pin(stream::select(receiver, runner)))) +} diff --git a/src/lib.rs b/src/lib.rs index f9f3952c41..47766e6fd1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -190,7 +190,6 @@ pub use crate::core::{ color, Alignment, Background, BorderRadius, Color, ContentFit, Degrees, Gradient, Length, Padding, Pixels, Point, Radians, Rectangle, Size, Vector, }; -pub use crate::runtime::Command; pub mod clipboard { //! Access the clipboard. @@ -239,6 +238,11 @@ pub mod mouse { }; } +pub mod command { + //! Run asynchronous actions. + pub use crate::runtime::command::{channel, Command}; +} + pub mod subscription { //! Listen to external events in your application. pub use iced_futures::subscription::{ @@ -287,6 +291,7 @@ pub mod widget { } pub use application::Application; +pub use command::Command; pub use error::Error; pub use event::Event; pub use executor::Executor;