Introduce `Task::abortable` :tada:

1 files changed, 37 insertions, 0 deletions

diff --git a/runtime/src/task.rs b/runtime/src/task.rs
index d1864473..b75aca89 100644
--- a/runtime/src/task.rs
+++ b/runtime/src/task.rs
@@ -159,6 +159,21 @@ impl<T> Task<T> {
         }
     }
 
+    /// Creates a new [`Task`] that can be aborted with the returned [`Handle`].
+    pub fn abortable(self) -> (Self, Handle)
+    where
+        T: 'static,
+    {
+        match self.0 {
+            Some(stream) => {
+                let (stream, handle) = stream::abortable(stream);
+
+                (Self(Some(boxed_stream(stream))), Handle(Some(handle)))
+            }
+            None => (Self(None), Handle(None)),
+        }
+    }
+
     /// Creates a new [`Task`] that runs the given [`Future`] and produces
     /// its output.
     pub fn future(future: impl Future<Output = T> + MaybeSend + 'static) -> Self
@@ -178,6 +193,28 @@ impl<T> Task<T> {
     }
 }
 
+/// A handle to a [`Task`] that can be used for aborting it.
+#[derive(Debug, Clone)]
+pub struct Handle(Option<stream::AbortHandle>);
+
+impl Handle {
+    /// Aborts the [`Task`] of this [`Handle`].
+    pub fn abort(&self) {
+        if let Some(handle) = &self.0 {
+            handle.abort();
+        }
+    }
+
+    /// Returns `true` if the [`Task`] of this [`Handle`] has been aborted.
+    pub fn is_aborted(&self) -> bool {
+        if let Some(handle) = &self.0 {
+            handle.is_aborted()
+        } else {
+            true
+        }
+    }
+}
+
 impl<T> Task<Option<T>> {
     /// Executes a new [`Task`] after this one, only when it produces `Some` value.
     ///