Handle view updates when following

Basic following now works. Editors' scroll positions
are their only replicated view state.

crates/editor/src/items.rs

@@ -64,7 +64,7 @@ impl FollowableItem for Editor {
     fn to_update_message(
         &self,
         event: &Self::Event,
-        cx: &AppContext,
+        _: &AppContext,
     ) -> Option<update_view::Variant> {
         match event {
             Event::ScrollPositionChanged => {

crates/server/src/rpc.rs

@@ -4252,7 +4252,7 @@ mod tests {
             })
             .await
             .unwrap();
-        let editor_a2 = workspace_a
+        let _editor_a2 = workspace_a
             .update(cx_a, |workspace, cx| {
                 workspace.open_path((worktree_id, "2.txt"), cx)
             })
@@ -4261,6 +4261,9 @@ mod tests {
 
         // Client B opens an editor.
         let workspace_b = client_b.build_workspace(&project_b, cx_b);
+        workspace_b.update(cx_b, |workspace, cx| {
+            workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
+        });
         let editor_b1 = workspace_b
             .update(cx_b, |workspace, cx| {
                 workspace.open_path((worktree_id, "1.txt"), cx)
@@ -4271,7 +4274,6 @@ mod tests {
         // Client B starts following client A.
         workspace_b
             .update(cx_b, |workspace, cx| {
-                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
                 let leader_id = *project_b.read(cx).collaborators().keys().next().unwrap();
                 workspace.follow(&leader_id.into(), cx).unwrap()
             })
@@ -4776,7 +4778,7 @@ mod tests {
             project: &ModelHandle<Project>,
             cx: &mut TestAppContext,
         ) -> ViewHandle<Workspace> {
-            let (window_id, _) = cx.add_window(|cx| EmptyView);
+            let (window_id, _) = cx.add_window(|_| EmptyView);
             cx.add_view(window_id, |cx| {
                 let fs = project.read(cx).fs().clone();
                 Workspace::new(

crates/workspace/src/pane.rs

@@ -1,7 +1,5 @@
 use super::{ItemHandle, SplitDirection};
 use crate::{Item, Settings, WeakItemHandle, Workspace};
-use anyhow::{anyhow, Result};
-use client::PeerId;
 use collections::{HashMap, VecDeque};
 use gpui::{
     action,
@@ -258,7 +256,7 @@ impl Pane {
                 let task = task.await;
                 if let Some(pane) = pane.upgrade(&cx) {
                     if let Some((project_entry_id, build_item)) = task.log_err() {
-                        pane.update(&mut cx, |pane, cx| {
+                        pane.update(&mut cx, |pane, _| {
                             pane.nav_history.borrow_mut().set_mode(mode);
                         });
                         let item = workspace.update(&mut cx, |workspace, cx| {

crates/workspace/src/workspace.rs

@@ -11,7 +11,7 @@ use client::{
     proto, Authenticate, ChannelList, Client, PeerId, Subscription, TypedEnvelope, User, UserStore,
 };
 use clock::ReplicaId;
-use collections::{HashMap, HashSet};
+use collections::{hash_map, HashMap, HashSet};
 use gpui::{
     action,
     color::Color,
@@ -37,6 +37,7 @@ pub use status_bar::StatusItemView;
 use std::{
     any::{Any, TypeId},
     cell::RefCell,
+    fmt,
     future::Future,
     path::{Path, PathBuf},
     rc::Rc,
@@ -298,7 +299,7 @@ impl<T: FollowableItem> FollowableItemHandle for ViewHandle<T> {
     }
 }
 
-pub trait ItemHandle: 'static {
+pub trait ItemHandle: 'static + fmt::Debug {
     fn tab_content(&self, style: &theme::Tab, cx: &AppContext) -> ElementBox;
     fn project_path(&self, cx: &AppContext) -> Option<ProjectPath>;
     fn project_entry_id(&self, cx: &AppContext) -> Option<ProjectEntryId>;
@@ -596,7 +597,7 @@ pub struct Workspace {
     status_bar: ViewHandle<StatusBar>,
     project: ModelHandle<Project>,
     leader_state: LeaderState,
-    follower_states_by_leader: HashMap<PeerId, HashMap<WeakViewHandle<Pane>, FollowerState>>,
+    follower_states_by_leader: HashMap<PeerId, FollowerState>,
     _observe_current_user: Task<()>,
 }
 
@@ -607,10 +608,12 @@ struct LeaderState {
 
 #[derive(Default)]
 struct FollowerState {
-    active_view_id: Option<usize>,
+    active_view_id: Option<u64>,
-    items_by_leader_view_id: HashMap<usize, FollowerItem>,
+    items_by_leader_view_id: HashMap<u64, FollowerItem>,
+    panes: HashSet<WeakViewHandle<Pane>>,
 }
 
+#[derive(Debug)]
 enum FollowerItem {
     Loading(Vec<proto::update_followers::update_view::Variant>),
     Loaded(Box<dyn FollowableItemHandle>),
@@ -1206,78 +1209,84 @@ impl Workspace {
             leader_id: leader_id.0,
         });
         Some(cx.spawn_weak(|this, mut cx| async move {
-            let mut response = request.await?;
+            let response = request.await?;
             if let Some(this) = this.upgrade(&cx) {
-                let mut item_tasks = Vec::new();
+                Self::add_views_from_leader(this.clone(), leader_id, response.views, &mut cx)
-                let (project, pane) = this.read_with(&cx, |this, _| {
+                    .await?;
-                    (this.project.clone(), this.active_pane().clone())
+                this.update(&mut cx, |this, cx| {
-                });
+                    this.follower_state(leader_id)?.active_view_id = response.active_view_id;
-                let item_builders = cx.update(|cx| {
+                    this.leader_updated(leader_id, cx);
-                    cx.default_global::<FollowableItemBuilders>()
+                    Ok::<_, anyhow::Error>(())
-                        .values()
+                })?;
-                        .map(|b| b.0)
+            }
-                        .collect::<Vec<_>>()
+            Ok(())
-                        .clone()
+        }))
-                });
+    }
-                for view in &mut response.views {
+
-                    let variant = view
+    async fn add_views_from_leader(
-                        .variant
+        this: ViewHandle<Self>,
-                        .take()
+        leader_id: PeerId,
-                        .ok_or_else(|| anyhow!("missing variant"))?;
+        views: Vec<proto::View>,
-                    cx.update(|cx| {
+        cx: &mut AsyncAppContext,
-                        let mut variant = Some(variant);
+    ) -> Result<()> {
-                        for build_item in &item_builders {
+        let (project, pane) = this.read_with(cx, |this, _| {
-                            if let Some(task) =
+            (this.project.clone(), this.active_pane().clone())
-                                build_item(pane.clone(), project.clone(), &mut variant, cx)
+        });
-                            {
+
-                                item_tasks.push(task);
+        let item_builders = cx.update(|cx| {
-                                break;
+            cx.default_global::<FollowableItemBuilders>()
-                            } else {
+                .values()
-                                assert!(variant.is_some());
+                .map(|b| b.0)
-                            }
+                .collect::<Vec<_>>()
-                        }
+                .clone()
-                    });
+        });
+
+        let mut item_tasks = Vec::new();
+        let mut leader_view_ids = Vec::new();
+        for view in views {
+            let mut variant = view.variant;
+            if variant.is_none() {
+                Err(anyhow!("missing variant"))?;
+            }
+            for build_item in &item_builders {
+                let task =
+                    cx.update(|cx| build_item(pane.clone(), project.clone(), &mut variant, cx));
+                if let Some(task) = task {
+                    item_tasks.push(task);
+                    leader_view_ids.push(view.id);
+                    break;
+                } else {
+                    assert!(variant.is_some());
                 }
+            }
+        }
 
-                this.update(&mut cx, |this, cx| {
+        let pane = pane.downgrade();
-                    this.follower_states_by_leader
+        let items = futures::future::try_join_all(item_tasks).await?;
-                        .entry(leader_id)
+        this.update(cx, |this, cx| {
-                        .or_default()
+            let state = this.follower_states_by_leader.entry(leader_id).or_default();
-                        .insert(
+            state.panes.insert(pane);
-                            pane.downgrade(),
+            for (id, item) in leader_view_ids.into_iter().zip(items) {
-                            FollowerState {
+                match state.items_by_leader_view_id.entry(id) {
-                                active_view_id: response.active_view_id.map(|id| id as usize),
+                    hash_map::Entry::Occupied(e) => {
-                                items_by_leader_view_id: Default::default(),
+                        let e = e.into_mut();
-                            },
+                        if let FollowerItem::Loading(updates) = e {
-                        );
+                            for update in updates.drain(..) {
-                });
-
-                let items = futures::future::try_join_all(item_tasks).await?;
-                this.update(&mut cx, |this, cx| {
-                    let follower_state = this
-                        .follower_states_by_leader
-                        .entry(leader_id)
-                        .or_default()
-                        .entry(pane.downgrade())
-                        .or_default();
-                    for (id, item) in response.views.iter().map(|v| v.id as usize).zip(items) {
-                        let prev_state = follower_state.items_by_leader_view_id.remove(&id);
-                        if let Some(FollowerItem::Loading(updates)) = prev_state {
-                            for update in updates {
                                 item.apply_update_message(update, cx)
                                     .context("failed to apply view update")
                                     .log_err();
                             }
                         }
-                        follower_state
+                        *e = FollowerItem::Loaded(item);
-                            .items_by_leader_view_id
-                            .insert(id, FollowerItem::Loaded(item));
                     }
-                    this.leader_updated(leader_id, cx);
+                    hash_map::Entry::Vacant(e) => {
-                });
+                        e.insert(FollowerItem::Loaded(item));
+                    }
+                }
             }
-            Ok(())
+        });
-        }))
+
+        Ok(())
     }
 
     fn update_followers(
@@ -1557,7 +1566,7 @@ impl Workspace {
         _: Arc<Client>,
         mut cx: AsyncAppContext,
     ) -> Result<()> {
-        this.update(&mut cx, |this, cx| {
+        this.update(&mut cx, |this, _| {
             this.leader_state
                 .followers
                 .remove(&envelope.original_sender_id()?);
@@ -1571,49 +1580,81 @@ impl Workspace {
         _: Arc<Client>,
         mut cx: AsyncAppContext,
     ) -> Result<()> {
-        this.update(&mut cx, |this, cx| {
+        let leader_id = envelope.original_sender_id()?;
-            let leader_id = envelope.original_sender_id()?;
+        match envelope
-            let follower_states = this
+            .payload
-                .follower_states_by_leader
+            .variant
-                .get_mut(&leader_id)
+            .ok_or_else(|| anyhow!("invalid update"))?
-                .ok_or_else(|| anyhow!("received follow update for an unfollowed peer"))?;
+        {
-            match envelope
+            proto::update_followers::Variant::UpdateActiveView(update_active_view) => {
-                .payload
+                this.update(&mut cx, |this, cx| {
-                .variant
+                    this.follower_state(leader_id)?.active_view_id = update_active_view.id;
-                .ok_or_else(|| anyhow!("invalid update"))?
+                    this.leader_updated(leader_id, cx);
-            {
+                    Ok::<_, anyhow::Error>(())
-                proto::update_followers::Variant::UpdateActiveView(update_active_view) => {
+                })
-                    for state in follower_states.values_mut() {
-                        state.active_view_id = update_active_view.id.map(|id| id as usize);
-                    }
-                }
-                proto::update_followers::Variant::UpdateView(update_view) => {
-                    for state in follower_states.values_mut() {
-                        // state.items_by_leader_view_id.get(k)
-                    }
-                }
-                proto::update_followers::Variant::CreateView(_) => todo!(),
             }
+            proto::update_followers::Variant::UpdateView(update_view) => {
+                this.update(&mut cx, |this, cx| {
+                    let variant = update_view
+                        .variant
+                        .ok_or_else(|| anyhow!("missing update view variant"))?;
+                    match this
+                        .follower_state(leader_id)?
+                        .items_by_leader_view_id
+                        .entry(update_view.id)
+                        .or_insert(FollowerItem::Loading(Vec::new()))
+                    {
+                        FollowerItem::Loaded(item) => {
+                            item.apply_update_message(variant, cx).log_err();
+                        }
+                        FollowerItem::Loading(updates) => updates.push(variant),
+                    }
+                    this.leader_updated(leader_id, cx);
+                    Ok(())
+                })
+            }
+            proto::update_followers::Variant::CreateView(view) => {
+                Self::add_views_from_leader(this.clone(), leader_id, vec![view], &mut cx).await?;
+                this.update(&mut cx, |this, cx| {
+                    this.leader_updated(leader_id, cx);
+                });
+                Ok(())
+            }
+        }
+        .log_err();
 
-            this.leader_updated(leader_id, cx);
+        Ok(())
-            Ok(())
+    }
-        })
+
+    fn follower_state(&mut self, leader_id: PeerId) -> Result<&mut FollowerState> {
+        self.follower_states_by_leader
+            .get_mut(&leader_id)
+            .ok_or_else(|| anyhow!("received follow update for an unfollowed peer"))
     }
 
     fn leader_updated(&mut self, leader_id: PeerId, cx: &mut ViewContext<Self>) -> Option<()> {
-        let mut items_to_add = Vec::new();
+        let state = self.follower_states_by_leader.get_mut(&leader_id)?;
-        for (pane, state) in self.follower_states_by_leader.get(&leader_id)? {
+        let active_item = state.items_by_leader_view_id.get(&state.active_view_id?)?;
-            if let Some((pane, active_view_id)) = pane.upgrade(cx).zip(state.active_view_id) {
+        if let FollowerItem::Loaded(item) = active_item {
-                if let Some(FollowerItem::Loaded(item)) =
+            let mut panes = Vec::new();
-                    state.items_by_leader_view_id.get(&active_view_id)
+            state.panes.retain(|pane| {
-                {
+                if let Some(pane) = pane.upgrade(cx) {
-                    items_to_add.push((pane, item.boxed_clone()));
+                    panes.push(pane);
+                    true
+                } else {
+                    false
                 }
-            }
+            });
-        }
 
-        for (pane, item) in items_to_add {
+            if panes.is_empty() {
-            Pane::add_item(self, pane, item.clone(), cx);
+                self.follower_states_by_leader.remove(&leader_id);
+            } else {
+                let item = item.boxed_clone();
+                for pane in panes {
+                    Pane::add_item(self, pane, item.clone(), cx);
+                }
+                cx.notify();
+            }
         }
 
         None