Merge pull request #2189 from zed-industries/labeled-tasks

Labeled Tasks

crates/gpui/src/app.rs

@@ -31,7 +31,7 @@ use uuid::Uuid;
 
 pub use action::*;
 use callback_collection::CallbackCollection;
-use collections::{hash_map::Entry, HashMap, HashSet, VecDeque};
+use collections::{hash_map::Entry, BTreeMap, HashMap, HashSet, VecDeque};
 pub use menu::*;
 use platform::Event;
 #[cfg(any(test, feature = "test-support"))]
@@ -474,6 +474,7 @@ type WindowBoundsCallback = Box<dyn FnMut(WindowBounds, Uuid, &mut MutableAppCon
 type KeystrokeCallback = Box<
     dyn FnMut(&Keystroke, &MatchResult, Option<&Box<dyn Action>>, &mut MutableAppContext) -> bool,
 >;
+type ActiveLabeledTasksCallback = Box<dyn FnMut(&mut MutableAppContext) -> bool>;
 type DeserializeActionCallback = fn(json: &str) -> anyhow::Result<Box<dyn Action>>;
 type WindowShouldCloseSubscriptionCallback = Box<dyn FnMut(&mut MutableAppContext) -> bool>;
 
@@ -503,6 +504,7 @@ pub struct MutableAppContext {
     window_fullscreen_observations: CallbackCollection<usize, WindowFullscreenCallback>,
     window_bounds_observations: CallbackCollection<usize, WindowBoundsCallback>,
     keystroke_observations: CallbackCollection<usize, KeystrokeCallback>,
+    active_labeled_task_observations: CallbackCollection<(), ActiveLabeledTasksCallback>,
 
     #[allow(clippy::type_complexity)]
     presenters_and_platform_windows:
@@ -514,6 +516,8 @@ pub struct MutableAppContext {
     pending_flushes: usize,
     flushing_effects: bool,
     halt_action_dispatch: bool,
+    next_labeled_task_id: usize,
+    active_labeled_tasks: BTreeMap<usize, &'static str>,
 }
 
 impl MutableAppContext {
@@ -562,6 +566,7 @@ impl MutableAppContext {
             window_bounds_observations: Default::default(),
             keystroke_observations: Default::default(),
             action_dispatch_observations: Default::default(),
+            active_labeled_task_observations: Default::default(),
             presenters_and_platform_windows: Default::default(),
             foreground,
             pending_effects: VecDeque::new(),
@@ -570,6 +575,8 @@ impl MutableAppContext {
             pending_flushes: 0,
             flushing_effects: false,
             halt_action_dispatch: false,
+            next_labeled_task_id: 0,
+            active_labeled_tasks: Default::default(),
         }
     }
 
@@ -794,6 +801,12 @@ impl MutableAppContext {
         window.screen().display_uuid()
     }
 
+    pub fn active_labeled_tasks<'a>(
+        &'a self,
+    ) -> impl DoubleEndedIterator<Item = &'static str> + 'a {
+        self.active_labeled_tasks.values().cloned()
+    }
+
     pub fn render_view(&mut self, params: RenderParams) -> Result<ElementBox> {
         let window_id = params.window_id;
         let view_id = params.view_id;
@@ -1160,6 +1173,19 @@ impl MutableAppContext {
         )
     }
 
+    pub fn observe_active_labeled_tasks<F>(&mut self, callback: F) -> Subscription
+    where
+        F: 'static + FnMut(&mut MutableAppContext) -> bool,
+    {
+        let subscription_id = post_inc(&mut self.next_subscription_id);
+        self.active_labeled_task_observations
+            .add_callback((), subscription_id, Box::new(callback));
+        Subscription::ActiveLabeledTasksObservation(
+            self.active_labeled_task_observations
+                .subscribe((), subscription_id),
+        )
+    }
+
     pub fn defer(&mut self, callback: impl 'static + FnOnce(&mut MutableAppContext)) {
         self.pending_effects.push_back(Effect::Deferred {
             callback: Box::new(callback),
@@ -2042,6 +2068,17 @@ impl MutableAppContext {
                             handled_by,
                             result,
                         } => self.handle_keystroke_effect(window_id, keystroke, handled_by, result),
+                        Effect::ActiveLabeledTasksChanged => {
+                            self.handle_active_labeled_tasks_changed_effect()
+                        }
+                        Effect::ActiveLabeledTasksObservation {
+                            subscription_id,
+                            callback,
+                        } => self.active_labeled_task_observations.add_callback(
+                            (),
+                            subscription_id,
+                            callback,
+                        ),
                     }
                     self.pending_notifications.clear();
                     self.remove_dropped_entities();
@@ -2449,26 +2486,68 @@ impl MutableAppContext {
         }
     }
 
+    fn handle_active_labeled_tasks_changed_effect(&mut self) {
+        self.active_labeled_task_observations
+            .clone()
+            .emit((), self, move |callback, this| {
+                callback(this);
+                true
+            });
+    }
+
     pub fn focus(&mut self, window_id: usize, view_id: Option<usize>) {
         self.pending_effects
             .push_back(Effect::Focus { window_id, view_id });
     }
 
-    pub fn spawn<F, Fut, T>(&self, f: F) -> Task<T>
+    fn spawn_internal<F, Fut, T>(&mut self, task_name: Option<&'static str>, f: F) -> Task<T>
     where
         F: FnOnce(AsyncAppContext) -> Fut,
         Fut: 'static + Future<Output = T>,
         T: 'static,
     {
+        let label_id = task_name.map(|task_name| {
+            let id = post_inc(&mut self.next_labeled_task_id);
+            self.active_labeled_tasks.insert(id, task_name);
+            self.pending_effects
+                .push_back(Effect::ActiveLabeledTasksChanged);
+            id
+        });
+
         let future = f(self.to_async());
         let cx = self.to_async();
         self.foreground.spawn(async move {
             let result = future.await;
-            cx.0.borrow_mut().flush_effects();
+            let mut cx = cx.0.borrow_mut();
+
+            if let Some(completed_label_id) = label_id {
+                cx.active_labeled_tasks.remove(&completed_label_id);
+                cx.pending_effects
+                    .push_back(Effect::ActiveLabeledTasksChanged);
+            }
+            cx.flush_effects();
             result
         })
     }
 
+    pub fn spawn_labeled<F, Fut, T>(&mut self, task_name: &'static str, f: F) -> Task<T>
+    where
+        F: FnOnce(AsyncAppContext) -> Fut,
+        Fut: 'static + Future<Output = T>,
+        T: 'static,
+    {
+        self.spawn_internal(Some(task_name), f)
+    }
+
+    pub fn spawn<F, Fut, T>(&mut self, f: F) -> Task<T>
+    where
+        F: FnOnce(AsyncAppContext) -> Fut,
+        Fut: 'static + Future<Output = T>,
+        T: 'static,
+    {
+        self.spawn_internal(None, f)
+    }
+
     pub fn to_async(&self) -> AsyncAppContext {
         AsyncAppContext(self.weak_self.as_ref().unwrap().upgrade().unwrap())
     }
@@ -2907,6 +2986,11 @@ pub enum Effect {
         window_id: usize,
         callback: WindowShouldCloseSubscriptionCallback,
     },
+    ActiveLabeledTasksChanged,
+    ActiveLabeledTasksObservation {
+        subscription_id: usize,
+        callback: ActiveLabeledTasksCallback,
+    },
 }
 
 impl Debug for Effect {
@@ -3066,6 +3150,16 @@ impl Debug for Effect {
                 )
                 .field("result", result)
                 .finish(),
+            Effect::ActiveLabeledTasksChanged => {
+                f.debug_struct("Effect::ActiveLabeledTasksChanged").finish()
+            }
+            Effect::ActiveLabeledTasksObservation {
+                subscription_id,
+                callback: _,
+            } => f
+                .debug_struct("Effect::ActiveLabeledTasksObservation")
+                .field("subscription_id", subscription_id)
+                .finish(),
         }
     }
 }
@@ -3480,7 +3574,7 @@ impl<'a, T: Entity> ModelContext<'a, T> {
         WeakModelHandle::new(self.model_id)
     }
 
-    pub fn spawn<F, Fut, S>(&self, f: F) -> Task<S>
+    pub fn spawn<F, Fut, S>(&mut self, f: F) -> Task<S>
     where
         F: FnOnce(ModelHandle<T>, AsyncAppContext) -> Fut,
         Fut: 'static + Future<Output = S>,
@@ -3490,7 +3584,7 @@ impl<'a, T: Entity> ModelContext<'a, T> {
         self.app.spawn(|cx| f(handle, cx))
     }
 
-    pub fn spawn_weak<F, Fut, S>(&self, f: F) -> Task<S>
+    pub fn spawn_weak<F, Fut, S>(&mut self, f: F) -> Task<S>
     where
         F: FnOnce(WeakModelHandle<T>, AsyncAppContext) -> Fut,
         Fut: 'static + Future<Output = S>,
@@ -3947,6 +4041,23 @@ impl<'a, T: View> ViewContext<'a, T> {
             })
     }
 
+    pub fn observe_active_labeled_tasks<F>(&mut self, mut callback: F) -> Subscription
+    where
+        F: 'static + FnMut(&mut T, &mut ViewContext<T>),
+    {
+        let observer = self.weak_handle();
+        self.app.observe_active_labeled_tasks(move |cx| {
+            if let Some(observer) = observer.upgrade(cx) {
+                observer.update(cx, |observer, cx| {
+                    callback(observer, cx);
+                });
+                true
+            } else {
+                false
+            }
+        })
+    }
+
     pub fn emit(&mut self, payload: T::Event) {
         self.app.pending_effects.push_back(Effect::Event {
             entity_id: self.view_id,
@@ -3993,7 +4104,17 @@ impl<'a, T: View> ViewContext<'a, T> {
         self.app.halt_action_dispatch = false;
     }
 
-    pub fn spawn<F, Fut, S>(&self, f: F) -> Task<S>
+    pub fn spawn_labeled<F, Fut, S>(&mut self, task_label: &'static str, f: F) -> Task<S>
+    where
+        F: FnOnce(ViewHandle<T>, AsyncAppContext) -> Fut,
+        Fut: 'static + Future<Output = S>,
+        S: 'static,
+    {
+        let handle = self.handle();
+        self.app.spawn_labeled(task_label, |cx| f(handle, cx))
+    }
+
+    pub fn spawn<F, Fut, S>(&mut self, f: F) -> Task<S>
     where
         F: FnOnce(ViewHandle<T>, AsyncAppContext) -> Fut,
         Fut: 'static + Future<Output = S>,
@@ -4003,7 +4124,7 @@ impl<'a, T: View> ViewContext<'a, T> {
         self.app.spawn(|cx| f(handle, cx))
     }
 
-    pub fn spawn_weak<F, Fut, S>(&self, f: F) -> Task<S>
+    pub fn spawn_weak<F, Fut, S>(&mut self, f: F) -> Task<S>
     where
         F: FnOnce(WeakViewHandle<T>, AsyncAppContext) -> Fut,
         Fut: 'static + Future<Output = S>,
@@ -5121,6 +5242,9 @@ pub enum Subscription {
     KeystrokeObservation(callback_collection::Subscription<usize, KeystrokeCallback>),
     ReleaseObservation(callback_collection::Subscription<usize, ReleaseObservationCallback>),
     ActionObservation(callback_collection::Subscription<(), ActionObservationCallback>),
+    ActiveLabeledTasksObservation(
+        callback_collection::Subscription<(), ActiveLabeledTasksCallback>,
+    ),
 }
 
 impl Subscription {
@@ -5137,6 +5261,7 @@ impl Subscription {
             Subscription::KeystrokeObservation(subscription) => subscription.id(),
             Subscription::ReleaseObservation(subscription) => subscription.id(),
             Subscription::ActionObservation(subscription) => subscription.id(),
+            Subscription::ActiveLabeledTasksObservation(subscription) => subscription.id(),
         }
     }
 
@@ -5153,6 +5278,7 @@ impl Subscription {
             Subscription::WindowBoundsObservation(subscription) => subscription.detach(),
             Subscription::ReleaseObservation(subscription) => subscription.detach(),
             Subscription::ActionObservation(subscription) => subscription.detach(),
+            Subscription::ActiveLabeledTasksObservation(subscription) => subscription.detach(),
         }
     }
 }
@@ -5161,6 +5287,7 @@ impl Subscription {
 mod tests {
     use super::*;
     use crate::{actions, elements::*, impl_actions, MouseButton, MouseButtonEvent};
+    use postage::{sink::Sink, stream::Stream};
     use serde::Deserialize;
     use smol::future::poll_once;
     use std::{
@@ -6776,6 +6903,26 @@ mod tests {
         assert_eq!(presenter.borrow().rendered_views.len(), 1);
     }
 
+    #[crate::test(self)]
+    async fn test_labeled_tasks(cx: &mut TestAppContext) {
+        assert_eq!(None, cx.update(|cx| cx.active_labeled_tasks().next()));
+        let (mut sender, mut reciever) = postage::oneshot::channel::<()>();
+        let task = cx
+            .update(|cx| cx.spawn_labeled("Test Label", |_| async move { reciever.recv().await }));
+
+        assert_eq!(
+            Some("Test Label"),
+            cx.update(|cx| cx.active_labeled_tasks().next())
+        );
+        sender
+            .send(())
+            .await
+            .expect("Could not send message to complete task");
+        task.await;
+
+        assert_eq!(None, cx.update(|cx| cx.active_labeled_tasks().next()));
+    }
+
     #[crate::test(self)]
     async fn test_window_activation(cx: &mut TestAppContext) {
         struct View(&'static str);