diff --git a/devtools/src/comet.rs b/devtools/src/comet.rs index 86d5998e1d..79d1a73905 100644 --- a/devtools/src/comet.rs +++ b/devtools/src/comet.rs @@ -1,11 +1,12 @@ -use crate::runtime::task::{self, Task}; +use crate::futures::futures::channel::mpsc; +use crate::runtime::task::Task; use std::process; pub const COMPATIBLE_REVISION: &str = "c4d45e3f502d9e18e0d9d4eda2c07093c62d8309"; pub fn launch() -> Task { - task::try_blocking(|mut sender| { + Task::blocking(|| { let cargo_install = process::Command::new("cargo") .args(["install", "--list"]) .output()?; @@ -37,7 +38,6 @@ pub fn launch() -> Task { .stderr(process::Stdio::null()) .spawn()?; - let _ = sender.try_send(()); return Ok(()); } @@ -46,48 +46,56 @@ pub fn launch() -> Task { } pub fn install() -> Task { - task::try_blocking(|mut sender| { - use std::io::{BufRead, BufReader}; - use std::process::{Command, Stdio}; - - let mut install = Command::new("cargo") - .args([ - "install", - "--locked", - "--git", - "https://github.com/iced-rs/comet.git", - "--rev", - COMPATIBLE_REVISION, - ]) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::piped()) - .spawn()?; - - let mut stderr = BufReader::new(install.stderr.take().expect("stderr must be piped")); - - let mut log = String::new(); - - while let Ok(n) = stderr.read_line(&mut log) { - if n == 0 { - let status = install.wait()?; - - if status.success() { - break; - } else { - return Err(install::Error::ProcessFailed(status)); + let (mut sender, receiver) = mpsc::channel(1); + + Task::batch([ + Task::stream(receiver).map(Ok), + Task::blocking(move || { + use std::io::{BufRead, BufReader}; + use std::process::{Command, Stdio}; + + let mut install = Command::new("cargo") + .args([ + "install", + "--locked", + "--git", + "https://github.com/iced-rs/comet.git", + "--rev", + COMPATIBLE_REVISION, + ]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn()?; + + let mut stderr = BufReader::new(install.stderr.take().expect("stderr must be piped")); + + let mut log = String::new(); + + while let Ok(n) = stderr.read_line(&mut log) { + if n == 0 { + let status = install.wait()?; + + if status.success() { + break; + } else { + return Err(install::Error::ProcessFailed(status)); + } } - } - let _ = sender.try_send(install::Event::Logged(log.trim_end().to_owned())); + let _ = sender.try_send(install::Event::Logged(log.trim_end().to_owned())); - log.clear(); - } + log.clear(); + } - let _ = sender.try_send(install::Event::Finished); + let _ = sender.try_send(install::Event::Finished); - Ok(()) - }) + Ok(()) + }) + .map(Result::err) + .and_then(Task::done) + .map(Err), + ]) } pub mod launch { diff --git a/devtools/src/lib.rs b/devtools/src/lib.rs index 2bbea4c0ad..a0d302dd83 100644 --- a/devtools/src/lib.rs +++ b/devtools/src/lib.rs @@ -23,7 +23,7 @@ use crate::core::{ use crate::futures::Subscription; use crate::program::Program; use crate::program::message; -use crate::runtime::task::{self, Task}; +use crate::runtime::task::Task; use crate::time_machine::TimeMachine; use crate::widget::{ bottom_right, button, center, column, container, opaque, row, scrollable, space, stack, text, @@ -159,11 +159,7 @@ where show_notification: true, time_machine: TimeMachine::new(), }, - Task::batch([task::blocking(|mut sender| { - thread::sleep(seconds(2)); - let _ = sender.try_send(()); - }) - .map(|_| Message::HideNotification)]), + Task::blocking(|| thread::sleep(seconds(2))).map(|_| Message::HideNotification), ) } diff --git a/runtime/src/task.rs b/runtime/src/task.rs index 5575bcf26b..f1552dab6e 100644 --- a/runtime/src/task.rs +++ b/runtime/src/task.rs @@ -265,6 +265,17 @@ impl Task { } } + /// Creates a new [`Task`] that runs the given closure in a new thread and + /// produces its output. + pub fn blocking(f: impl FnOnce() -> T + Send + 'static) -> Self + where + T: Send + 'static, + { + let (result_tx, result_rx) = oneshot::channel(); + let _ = thread::spawn(|| result_tx.send(f())); + Task::perform(result_rx, Result::ok).and_then(Task::done) + } + /// Returns the amount of work "units" of the [`Task`]. pub fn units(&self) -> usize { self.units @@ -467,50 +478,6 @@ pub fn into_stream(task: Task) -> Option>> { task.stream } -/// Creates a new [`Task`] that will run the given closure in a new thread. -/// -/// Any data sent by the closure through the [`mpsc::Sender`] will be produced -/// by the [`Task`]. -pub fn blocking(f: impl FnOnce(mpsc::Sender) + Send + 'static) -> Task -where - T: Send + 'static, -{ - let (sender, receiver) = mpsc::channel(1); - - let _ = thread::spawn(move || { - f(sender); - }); - - Task::stream(receiver) -} - -/// Creates a new [`Task`] that will run the given closure that can fail in a new -/// thread. -/// -/// Any data sent by the closure through the [`mpsc::Sender`] will be produced -/// by the [`Task`]. -pub fn try_blocking( - f: impl FnOnce(mpsc::Sender) -> Result<(), E> + Send + 'static, -) -> Task> -where - T: Send + 'static, - E: Send + 'static, -{ - let (sender, receiver) = mpsc::channel(1); - let (error_sender, error_receiver) = oneshot::channel(); - - let _ = thread::spawn(move || { - if let Err(error) = f(sender) { - let _ = error_sender.send(Err(error)); - } - }); - - Task::stream(stream::select( - receiver.map(Ok), - stream::once(error_receiver).filter_map(async |result| result.ok()), - )) -} - async fn yield_now() { struct YieldNow { yielded: bool, diff --git a/tester/src/lib.rs b/tester/src/lib.rs index 1ce6635320..0b18331cbc 100644 --- a/tester/src/lib.rs +++ b/tester/src/lib.rs @@ -21,7 +21,7 @@ use crate::core::window; use crate::core::{Color, Element, Font, Settings, Size, Theme}; use crate::futures::futures::channel::mpsc; use crate::program::Program; -use crate::runtime::task::{self, Task}; +use crate::runtime::task::Task; use crate::test::emulator; use crate::test::ice; use crate::test::instruction; @@ -298,10 +298,8 @@ impl Tester

{ Task::future(import) .and_then(|file| { - task::blocking(move |mut sender| { - let _ = sender.try_send(Ice::parse( - &fs::read_to_string(file.path()).unwrap_or_default(), - )); + Task::blocking(move || { + Ice::parse(&fs::read_to_string(file.path()).unwrap_or_default()) }) }) .map(Event::Imported)