Add Task::blocking helper

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<launch::Result> {
-    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<launch::Result> {
                 .stderr(process::Stdio::null())
                 .spawn()?;
 
-            let _ = sender.try_send(());
             return Ok(());
         }
 
@@ -46,48 +46,56 @@ pub fn launch() -> Task<launch::Result> {
 }
 
 pub fn install() -> Task<install::Result> {
-    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 {

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),
         )
     }
 

runtime/src/task.rs

@@ -265,6 +265,17 @@ impl<T> Task<T> {
         }
     }
 
+    /// 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<T>(task: Task<T>) -> Option<BoxStream<Action<T>>> {
     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<T>(f: impl FnOnce(mpsc::Sender<T>) + Send + 'static) -> Task<T>
-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<T, E>(
-    f: impl FnOnce(mpsc::Sender<T>) -> Result<(), E> + Send + 'static,
-) -> Task<Result<T, E>>
-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,

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<P: Program + 'static> Tester<P> {
 
                 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)