From ad37034960974a7a70649b64b2e244ea64bbadfe Mon Sep 17 00:00:00 2001 From: Max Brunsfeld Date: Thu, 15 Dec 2022 17:19:32 -0800 Subject: [PATCH 1/5] Identify LiveKit room participants by user id, not peer id This way, their participant id can remain the same when they reconnect. --- crates/call/src/participant.rs | 1 + crates/call/src/room.rs | 41 +++++++++++++------- crates/collab/src/integration_tests.rs | 2 +- crates/collab/src/rpc.rs | 6 +-- crates/collab_ui/src/collab_titlebar_item.rs | 17 ++++---- crates/collab_ui/src/collab_ui.rs | 2 +- crates/collab_ui/src/contact_list.rs | 22 +++++------ crates/workspace/src/pane_group.rs | 2 +- crates/workspace/src/workspace.rs | 4 +- 9 files changed, 55 insertions(+), 42 deletions(-) diff --git a/crates/call/src/participant.rs b/crates/call/src/participant.rs index d5c6d85154bd796c6624a42e938b4501eaa3f26c..4b9e7ba034fb23d118cb5963d335e5c16201d590 100644 --- a/crates/call/src/participant.rs +++ b/crates/call/src/participant.rs @@ -39,6 +39,7 @@ pub struct LocalParticipant { #[derive(Clone, Debug)] pub struct RemoteParticipant { pub user: Arc, + pub peer_id: proto::PeerId, pub projects: Vec, pub location: ParticipantLocation, pub tracks: HashMap>, diff --git a/crates/call/src/room.rs b/crates/call/src/room.rs index 1d22fe50f1a8ff4e9ac530ff991f5f9010a2b4e6..f2d40b012900c91fd1c3c3b769f62a355da29736 100644 --- a/crates/call/src/room.rs +++ b/crates/call/src/room.rs @@ -3,7 +3,10 @@ use crate::{ IncomingCall, }; use anyhow::{anyhow, Result}; -use client::{proto, Client, TypedEnvelope, User, UserStore}; +use client::{ + proto::{self, PeerId}, + Client, TypedEnvelope, User, UserStore, +}; use collections::{BTreeMap, HashSet}; use futures::{FutureExt, StreamExt}; use gpui::{ @@ -41,7 +44,7 @@ pub struct Room { live_kit: Option, status: RoomStatus, local_participant: LocalParticipant, - remote_participants: BTreeMap, + remote_participants: BTreeMap, pending_participants: Vec>, participant_user_ids: HashSet, pending_call_count: usize, @@ -349,10 +352,16 @@ impl Room { &self.local_participant } - pub fn remote_participants(&self) -> &BTreeMap { + pub fn remote_participants(&self) -> &BTreeMap { &self.remote_participants } + pub fn remote_participant_for_peer_id(&self, peer_id: PeerId) -> Option<&RemoteParticipant> { + self.remote_participants + .values() + .find(|p| p.peer_id == peer_id) + } + pub fn pending_participants(&self) -> &[Arc] { &self.pending_participants } @@ -421,11 +430,11 @@ impl Room { for (participant, user) in room.participants.into_iter().zip(participants) { let Some(peer_id) = participant.peer_id else { continue }; this.participant_user_ids.insert(participant.user_id); - participant_peer_ids.insert(peer_id); + participant_peer_ids.insert(participant.user_id); let old_projects = this .remote_participants - .get(&peer_id) + .get(&participant.user_id) .into_iter() .flat_map(|existing| &existing.projects) .map(|project| project.id) @@ -454,7 +463,8 @@ impl Room { let location = ParticipantLocation::from_proto(participant.location) .unwrap_or(ParticipantLocation::External); - if let Some(remote_participant) = this.remote_participants.get_mut(&peer_id) + if let Some(remote_participant) = + this.remote_participants.get_mut(&participant.user_id) { remote_participant.projects = participant.projects; if location != remote_participant.location { @@ -465,9 +475,10 @@ impl Room { } } else { this.remote_participants.insert( - peer_id, + participant.user_id, RemoteParticipant { user: user.clone(), + peer_id, projects: participant.projects, location, tracks: Default::default(), @@ -488,8 +499,8 @@ impl Room { } } - this.remote_participants.retain(|peer_id, participant| { - if participant_peer_ids.contains(peer_id) { + this.remote_participants.retain(|user_id, participant| { + if participant_peer_ids.contains(user_id) { true } else { for project in &participant.projects { @@ -531,11 +542,11 @@ impl Room { ) -> Result<()> { match change { RemoteVideoTrackUpdate::Subscribed(track) => { - let peer_id = track.publisher_id().parse()?; + let user_id = track.publisher_id().parse()?; let track_id = track.sid().to_string(); let participant = self .remote_participants - .get_mut(&peer_id) + .get_mut(&user_id) .ok_or_else(|| anyhow!("subscribed to track by unknown participant"))?; participant.tracks.insert( track_id.clone(), @@ -544,21 +555,21 @@ impl Room { }), ); cx.emit(Event::RemoteVideoTracksChanged { - participant_id: peer_id, + participant_id: participant.peer_id, }); } RemoteVideoTrackUpdate::Unsubscribed { publisher_id, track_id, } => { - let peer_id = publisher_id.parse()?; + let user_id = publisher_id.parse()?; let participant = self .remote_participants - .get_mut(&peer_id) + .get_mut(&user_id) .ok_or_else(|| anyhow!("unsubscribed from track by unknown participant"))?; participant.tracks.remove(&track_id); cx.emit(Event::RemoteVideoTracksChanged { - participant_id: peer_id, + participant_id: participant.peer_id, }); } } diff --git a/crates/collab/src/integration_tests.rs b/crates/collab/src/integration_tests.rs index 2629d340dac0e745e1027d3cd17913023fcb45d3..6c773503bbfac607f91a5b9eefaadafab9c67a21 100644 --- a/crates/collab/src/integration_tests.rs +++ b/crates/collab/src/integration_tests.rs @@ -199,7 +199,7 @@ async fn test_basic_calls( assert_eq!(participant_id, client_a.peer_id().unwrap()); room_b.read_with(cx_b, |room, _| { assert_eq!( - room.remote_participants()[&client_a.peer_id().unwrap()] + room.remote_participants()[&client_a.user_id().unwrap()] .tracks .len(), 1 diff --git a/crates/collab/src/rpc.rs b/crates/collab/src/rpc.rs index 6a8ae61ed0b7bff3ad44741a2a21528dbb32a5b7..03e6eb50e2ab1d80b02d287a85448fd30d0cdb62 100644 --- a/crates/collab/src/rpc.rs +++ b/crates/collab/src/rpc.rs @@ -850,7 +850,7 @@ async fn create_room( .trace_err() { if let Some(token) = live_kit - .room_token(&live_kit_room, &session.connection_id.to_string()) + .room_token(&live_kit_room, &session.user_id.to_string()) .trace_err() { Some(proto::LiveKitConnectionInfo { @@ -918,7 +918,7 @@ async fn join_room( let live_kit_connection_info = if let Some(live_kit) = session.live_kit_client.as_ref() { if let Some(token) = live_kit - .room_token(&room.live_kit_room, &session.connection_id.to_string()) + .room_token(&room.live_kit_room, &session.user_id.to_string()) .trace_err() { Some(proto::LiveKitConnectionInfo { @@ -2066,7 +2066,7 @@ async fn leave_room_for_session(session: &Session) -> Result<()> { if let Some(live_kit) = session.live_kit_client.as_ref() { live_kit - .remove_participant(live_kit_room.clone(), session.connection_id.to_string()) + .remove_participant(live_kit_room.clone(), session.user_id.to_string()) .await .trace_err(); diff --git a/crates/collab_ui/src/collab_titlebar_item.rs b/crates/collab_ui/src/collab_titlebar_item.rs index 2288f77cd358996f5e0ecba98d09d56a42a3dd76..3351fb9eb9c14c7f82f0f4d27243776a5b419161 100644 --- a/crates/collab_ui/src/collab_titlebar_item.rs +++ b/crates/collab_ui/src/collab_titlebar_item.rs @@ -342,24 +342,27 @@ impl CollabTitlebarItem { let mut participants = room .read(cx) .remote_participants() - .iter() - .map(|(peer_id, collaborator)| (*peer_id, collaborator.clone())) + .values() + .cloned() .collect::>(); - participants - .sort_by_key(|(peer_id, _)| Some(project.collaborators().get(peer_id)?.replica_id)); + participants.sort_by_key(|p| Some(project.collaborators().get(&p.peer_id)?.replica_id)); participants .into_iter() - .filter_map(|(peer_id, participant)| { + .filter_map(|participant| { let project = workspace.read(cx).project().read(cx); let replica_id = project .collaborators() - .get(&peer_id) + .get(&participant.peer_id) .map(|collaborator| collaborator.replica_id); let user = participant.user.clone(); Some(self.render_avatar( &user, replica_id, - Some((peer_id, &user.github_login, participant.location)), + Some(( + participant.peer_id, + &user.github_login, + participant.location, + )), workspace, theme, cx, diff --git a/crates/collab_ui/src/collab_ui.rs b/crates/collab_ui/src/collab_ui.rs index 1b851c3f7595e000e3242bec529b9ff4c8d7bc45..1041382515bfb398c6f4dbe7ecc674c25e71aa42 100644 --- a/crates/collab_ui/src/collab_ui.rs +++ b/crates/collab_ui/src/collab_ui.rs @@ -73,7 +73,7 @@ pub fn init(app_state: Arc, cx: &mut MutableAppContext) { .remote_participants() .iter() .find(|(_, participant)| participant.user.id == follow_user_id) - .map(|(peer_id, _)| *peer_id) + .map(|(_, p)| p.peer_id) .or_else(|| { // If we couldn't follow the given user, follow the host instead. let collaborator = workspace diff --git a/crates/collab_ui/src/contact_list.rs b/crates/collab_ui/src/contact_list.rs index 48a4d1a2b541d69f7fa8c41ab21da1e766103301..ba6b236a826b09c956b17351d2a2fc64a77e917e 100644 --- a/crates/collab_ui/src/contact_list.rs +++ b/crates/collab_ui/src/contact_list.rs @@ -461,15 +461,13 @@ impl ContactList { // Populate remote participants. self.match_candidates.clear(); self.match_candidates - .extend( - room.remote_participants() - .iter() - .map(|(peer_id, participant)| StringMatchCandidate { - id: peer_id.as_u64() as usize, - string: participant.user.github_login.clone(), - char_bag: participant.user.github_login.chars().collect(), - }), - ); + .extend(room.remote_participants().iter().map(|(_, participant)| { + StringMatchCandidate { + id: participant.user.id as usize, + string: participant.user.github_login.clone(), + char_bag: participant.user.github_login.chars().collect(), + } + })); let matches = executor.block(match_strings( &self.match_candidates, &query, @@ -479,8 +477,8 @@ impl ContactList { executor.clone(), )); for mat in matches { - let peer_id = PeerId::from_u64(mat.candidate_id as u64); - let participant = &room.remote_participants()[&peer_id]; + let user_id = mat.candidate_id as u64; + let participant = &room.remote_participants()[&user_id]; participant_entries.push(ContactEntry::CallParticipant { user: participant.user.clone(), is_pending: false, @@ -496,7 +494,7 @@ impl ContactList { } if !participant.tracks.is_empty() { participant_entries.push(ContactEntry::ParticipantScreen { - peer_id, + peer_id: participant.peer_id, is_last: true, }); } diff --git a/crates/workspace/src/pane_group.rs b/crates/workspace/src/pane_group.rs index b8e73d6f6f3c3f7dec6cb023be5dbf14f4facf1c..72e4baee2e4644ec7f91551f585b62c13178506d 100644 --- a/crates/workspace/src/pane_group.rs +++ b/crates/workspace/src/pane_group.rs @@ -148,7 +148,7 @@ impl Member { .and_then(|leader_id| { let room = active_call?.read(cx).room()?.read(cx); let collaborator = project.read(cx).collaborators().get(leader_id)?; - let participant = room.remote_participants().get(&leader_id)?; + let participant = room.remote_participant_for_peer_id(*leader_id)?; Some((collaborator.replica_id, participant)) }); diff --git a/crates/workspace/src/workspace.rs b/crates/workspace/src/workspace.rs index c30dc2ea2997ff25a9dfe7287f40cb2d70a85a90..287bbc2b1325e4ddaa520e6bfe59286b6d76941c 100644 --- a/crates/workspace/src/workspace.rs +++ b/crates/workspace/src/workspace.rs @@ -2141,7 +2141,7 @@ impl Workspace { let call = self.active_call()?; let room = call.read(cx).room()?.read(cx); - let participant = room.remote_participants().get(&leader_id)?; + let participant = room.remote_participant_for_peer_id(leader_id)?; let mut items_to_add = Vec::new(); match participant.location { @@ -2190,7 +2190,7 @@ impl Workspace { ) -> Option> { let call = self.active_call()?; let room = call.read(cx).room()?.read(cx); - let participant = room.remote_participants().get(&peer_id)?; + let participant = room.remote_participant_for_peer_id(peer_id)?; let track = participant.tracks.values().next()?.clone(); let user = participant.user.clone(); From aa44de3d1674ef860af337f4f55fbdcb19fd09e0 Mon Sep 17 00:00:00 2001 From: Antonio Scandurra Date: Fri, 16 Dec 2022 10:52:32 +0100 Subject: [PATCH 2/5] Fix test ensuring room is left when disconnected from LiveKit --- crates/call/src/room.rs | 5 ++--- crates/collab/src/integration_tests.rs | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/call/src/room.rs b/crates/call/src/room.rs index f2d40b012900c91fd1c3c3b769f62a355da29736..8e2c38b3f8db37965bd8dce862fd2473db8a6c71 100644 --- a/crates/call/src/room.rs +++ b/crates/call/src/room.rs @@ -426,11 +426,9 @@ impl Room { } if let Some(participants) = remote_participants.log_err() { - let mut participant_peer_ids = HashSet::default(); for (participant, user) in room.participants.into_iter().zip(participants) { let Some(peer_id) = participant.peer_id else { continue }; this.participant_user_ids.insert(participant.user_id); - participant_peer_ids.insert(participant.user_id); let old_projects = this .remote_participants @@ -467,6 +465,7 @@ impl Room { this.remote_participants.get_mut(&participant.user_id) { remote_participant.projects = participant.projects; + remote_participant.peer_id = peer_id; if location != remote_participant.location { remote_participant.location = location; cx.emit(Event::ParticipantLocationChanged { @@ -500,7 +499,7 @@ impl Room { } this.remote_participants.retain(|user_id, participant| { - if participant_peer_ids.contains(user_id) { + if this.participant_user_ids.contains(user_id) { true } else { for project in &participant.projects { diff --git a/crates/collab/src/integration_tests.rs b/crates/collab/src/integration_tests.rs index 6c773503bbfac607f91a5b9eefaadafab9c67a21..0c26486667fb82f606d1e1164e3ded6963001b74 100644 --- a/crates/collab/src/integration_tests.rs +++ b/crates/collab/src/integration_tests.rs @@ -492,7 +492,7 @@ async fn test_client_disconnecting_from_room( // to automatically leave the room. server .test_live_kit_server - .disconnect_client(client_b.peer_id().unwrap().to_string()) + .disconnect_client(client_b.user_id().unwrap().to_string()) .await; deterministic.run_until_parked(); active_call_a.update(cx_a, |call, _| assert!(call.room().is_none())); From 21ab1bb4343c99e11ba45b3b84fbc9e91d12f127 Mon Sep 17 00:00:00 2001 From: Antonio Scandurra Date: Fri, 16 Dec 2022 11:45:42 +0100 Subject: [PATCH 3/5] Remove unnecessary `PeerId` parsing code --- crates/rpc/src/proto.rs | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/crates/rpc/src/proto.rs b/crates/rpc/src/proto.rs index 385caf3565f639b973007a67669378f59584ea55..f833db514dd9918d1b58e6d93292781800529a46 100644 --- a/crates/rpc/src/proto.rs +++ b/crates/rpc/src/proto.rs @@ -6,7 +6,6 @@ use prost::Message as _; use serde::Serialize; use std::any::{Any, TypeId}; use std::fmt; -use std::str::FromStr; use std::{ cmp, fmt::Debug, @@ -119,23 +118,6 @@ impl fmt::Display for PeerId { } } -impl FromStr for PeerId { - type Err = anyhow::Error; - - fn from_str(s: &str) -> Result { - let mut components = s.split('/'); - let owner_id = components - .next() - .ok_or_else(|| anyhow!("invalid peer id {:?}", s))? - .parse()?; - let id = components - .next() - .ok_or_else(|| anyhow!("invalid peer id {:?}", s))? - .parse()?; - Ok(PeerId { owner_id, id }) - } -} - messages!( (Ack, Foreground), (AddProjectCollaborator, Foreground), From 457e1046c88660a6ab5dbd29842077070371ce77 Mon Sep 17 00:00:00 2001 From: Antonio Scandurra Date: Fri, 16 Dec 2022 11:48:14 +0100 Subject: [PATCH 4/5] Bump protocol version --- crates/rpc/src/rpc.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rpc/src/rpc.rs b/crates/rpc/src/rpc.rs index d06f1c1c5ce0c1e54ea1a5d85c0b8acb8edb5f23..01dda55cefc1d4193c5fee20a59f69a494c93d5c 100644 --- a/crates/rpc/src/rpc.rs +++ b/crates/rpc/src/rpc.rs @@ -6,4 +6,4 @@ pub use conn::Connection; pub use peer::*; mod macros; -pub const PROTOCOL_VERSION: u32 = 42; +pub const PROTOCOL_VERSION: u32 = 43; From a5f624203e0af4b5926fd8429eca8ff6ca3fdef5 Mon Sep 17 00:00:00 2001 From: Antonio Scandurra Date: Fri, 16 Dec 2022 12:02:03 +0100 Subject: [PATCH 5/5] collab 0.4.1 --- Cargo.lock | 2 +- crates/collab/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 85df40e9b08b71c9efa3fa4980fb26996b409a5f..5063e35c705fda2846b30ea5156515bc7a1abe11 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1131,7 +1131,7 @@ dependencies = [ [[package]] name = "collab" -version = "0.4.0" +version = "0.4.1" dependencies = [ "anyhow", "async-tungstenite", diff --git a/crates/collab/Cargo.toml b/crates/collab/Cargo.toml index f9e9194496a290fe66297dafc3503ac6ef6f14bb..2bdc1f36a12220255711f6fc1b3333df6e89b8ea 100644 --- a/crates/collab/Cargo.toml +++ b/crates/collab/Cargo.toml @@ -3,7 +3,7 @@ authors = ["Nathan Sobo "] default-run = "collab" edition = "2021" name = "collab" -version = "0.4.0" +version = "0.4.1" [[bin]] name = "collab"