Antonio Scandurra created
crates/call/src/participant.rs | 4 -
crates/call/src/room.rs | 34 ---------------
crates/collab/src/rpc.rs | 46 ++++++++++++++++-----
crates/collab/src/rpc/store.rs | 78 +++++++++++++++++++++++++++++------
crates/project/src/project.rs | 4
crates/rpc/proto/zed.proto | 4 +
6 files changed, 105 insertions(+), 65 deletions(-)
@@ -20,10 +20,6 @@ impl ParticipantLocation {
}
}
-pub struct LocalParticipant {
- pub projects: Vec<ModelHandle<Project>>,
-}
-
pub struct RemoteParticipant {
pub user_id: u64,
pub projects: Vec<ModelHandle<Project>>,
@@ -1,10 +1,9 @@
-use crate::participant::{LocalParticipant, ParticipantLocation, RemoteParticipant};
+use crate::participant::{ParticipantLocation, RemoteParticipant};
use anyhow::{anyhow, Result};
use client::{incoming_call::IncomingCall, proto, Client, PeerId, TypedEnvelope, User, UserStore};
use collections::HashMap;
use futures::StreamExt;
use gpui::{AsyncAppContext, Entity, ModelContext, ModelHandle, MutableAppContext, Task};
-use project::Project;
use std::sync::Arc;
use util::ResultExt;
@@ -15,7 +14,6 @@ pub enum Event {
pub struct Room {
id: u64,
status: RoomStatus,
- local_participant: LocalParticipant,
remote_participants: HashMap<PeerId, RemoteParticipant>,
pending_users: Vec<Arc<User>>,
client: Arc<Client>,
@@ -53,9 +51,6 @@ impl Room {
Self {
id,
status: RoomStatus::Online,
- local_participant: LocalParticipant {
- projects: Default::default(),
- },
remote_participants: Default::default(),
pending_users: Default::default(),
_subscriptions: vec![client.add_message_handler(cx.handle(), Self::handle_room_updated)],
@@ -179,33 +174,6 @@ impl Room {
Ok(())
})
}
-
- pub fn publish_project(&mut self, project: ModelHandle<Project>) -> Task<Result<()>> {
- if self.status.is_offline() {
- return Task::ready(Err(anyhow!("room is offline")));
- }
-
- todo!()
- }
-
- pub fn unpublish_project(&mut self, project: ModelHandle<Project>) -> Task<Result<()>> {
- if self.status.is_offline() {
- return Task::ready(Err(anyhow!("room is offline")));
- }
-
- todo!()
- }
-
- pub fn set_active_project(
- &mut self,
- project: Option<&ModelHandle<Project>>,
- ) -> Task<Result<()>> {
- if self.status.is_offline() {
- return Task::ready(Err(anyhow!("room is offline")));
- }
-
- todo!()
- }
}
#[derive(Copy, Clone, PartialEq, Eq)]
@@ -611,8 +611,34 @@ impl Server {
async fn leave_room(self: Arc<Server>, message: TypedEnvelope<proto::LeaveRoom>) -> Result<()> {
let room_id = message.payload.id;
let mut store = self.store().await;
- let room = store.leave_room(room_id, message.sender_id)?;
- self.room_updated(room);
+ let left_room = store.leave_room(room_id, message.sender_id)?;
+
+ for project in left_room.unshared_projects {
+ for connection_id in project.connection_ids() {
+ self.peer.send(
+ connection_id,
+ proto::UnshareProject {
+ project_id: project.id.to_proto(),
+ },
+ )?;
+ }
+ }
+
+ for project in left_room.left_projects {
+ if project.remove_collaborator {
+ for connection_id in project.connection_ids {
+ self.peer.send(
+ connection_id,
+ proto::RemoveProjectCollaborator {
+ project_id: project.id.to_proto(),
+ peer_id: message.sender_id.0,
+ },
+ )?;
+ }
+ }
+ }
+
+ self.room_updated(left_room.room);
Ok(())
}
@@ -696,13 +722,12 @@ impl Server {
.await
.user_id_for_connection(request.sender_id)?;
let project_id = self.app_state.db.register_project(user_id).await?;
- self.store()
- .await
- .share_project(request.sender_id, project_id)?;
-
+ let mut store = self.store().await;
+ let room = store.share_project(request.payload.room_id, project_id, request.sender_id)?;
response.send(proto::ShareProjectResponse {
project_id: project_id.to_proto(),
})?;
+ self.room_updated(room);
Ok(())
}
@@ -712,15 +737,14 @@ impl Server {
message: TypedEnvelope<proto::UnshareProject>,
) -> Result<()> {
let project_id = ProjectId::from_proto(message.payload.project_id);
- let project = self
- .store()
- .await
- .unshare_project(project_id, message.sender_id)?;
+ let mut store = self.store().await;
+ let (room, project) = store.unshare_project(project_id, message.sender_id)?;
broadcast(
message.sender_id,
project.guest_connection_ids(),
|conn_id| self.peer.send(conn_id, message.payload.clone()),
);
+ self.room_updated(room);
Ok(())
}
@@ -882,7 +906,7 @@ impl Server {
let project;
{
let mut store = self.store().await;
- project = store.leave_project(sender_id, project_id)?;
+ project = store.leave_project(project_id, sender_id)?;
tracing::info!(
%project_id,
host_user_id = %project.host_user_id,
@@ -44,6 +44,8 @@ pub struct Call {
#[derive(Serialize)]
pub struct Project {
+ pub id: ProjectId,
+ pub room_id: RoomId,
pub host_connection_id: ConnectionId,
pub host: Collaborator,
pub guests: HashMap<ConnectionId, Collaborator>,
@@ -90,12 +92,19 @@ pub struct RemovedConnectionState {
}
pub struct LeftProject {
+ pub id: ProjectId,
pub host_user_id: UserId,
pub host_connection_id: ConnectionId,
pub connection_ids: Vec<ConnectionId>,
pub remove_collaborator: bool,
}
+pub struct LeftRoom<'a> {
+ pub room: &'a proto::Room,
+ pub unshared_projects: Vec<Project>,
+ pub left_projects: Vec<LeftProject>,
+}
+
#[derive(Copy, Clone)]
pub struct Metrics {
pub connections: usize,
@@ -199,9 +208,9 @@ impl Store {
// Unshare and leave all projects.
for project_id in connection_projects {
- if let Ok(project) = self.unshare_project(project_id, connection_id) {
+ if let Ok((_, project)) = self.unshare_project(project_id, connection_id) {
result.hosted_projects.insert(project_id, project);
- } else if self.leave_project(connection_id, project_id).is_ok() {
+ } else if self.leave_project(project_id, connection_id).is_ok() {
result.guest_project_ids.insert(project_id);
}
}
@@ -424,11 +433,7 @@ impl Store {
Ok((room, recipient_connection_ids))
}
- pub fn leave_room(
- &mut self,
- room_id: RoomId,
- connection_id: ConnectionId,
- ) -> Result<&proto::Room> {
+ pub fn leave_room(&mut self, room_id: RoomId, connection_id: ConnectionId) -> Result<LeftRoom> {
let connection = self
.connections
.get_mut(&connection_id)
@@ -454,7 +459,22 @@ impl Store {
.retain(|participant| participant.peer_id != connection_id.0);
connected_user.active_call = None;
- Ok(room)
+ let mut unshared_projects = Vec::new();
+ let mut left_projects = Vec::new();
+ for project_id in connection.projects.clone() {
+ if let Ok((_, project)) = self.unshare_project(project_id, connection_id) {
+ unshared_projects.push(project);
+ } else if let Ok(project) = self.leave_project(project_id, connection_id) {
+ left_projects.push(project);
+ }
+ }
+
+ let room = self.rooms.get(&room_id).unwrap();
+ Ok(LeftRoom {
+ room,
+ unshared_projects,
+ left_projects,
+ })
}
pub fn room(&self, room_id: RoomId) -> Option<&proto::Room> {
@@ -564,17 +584,32 @@ impl Store {
pub fn share_project(
&mut self,
- host_connection_id: ConnectionId,
+ room_id: RoomId,
project_id: ProjectId,
- ) -> Result<()> {
+ host_connection_id: ConnectionId,
+ ) -> Result<&proto::Room> {
let connection = self
.connections
.get_mut(&host_connection_id)
.ok_or_else(|| anyhow!("no such connection"))?;
+
+ let room = self
+ .rooms
+ .get_mut(&room_id)
+ .ok_or_else(|| anyhow!("no such room"))?;
+ let participant = room
+ .participants
+ .iter_mut()
+ .find(|participant| participant.peer_id == host_connection_id.0)
+ .ok_or_else(|| anyhow!("no such room"))?;
+ participant.project_ids.push(project_id.to_proto());
+
connection.projects.insert(project_id);
self.projects.insert(
project_id,
Project {
+ id: project_id,
+ room_id,
host_connection_id,
host: Collaborator {
user_id: connection.user_id,
@@ -588,14 +623,15 @@ impl Store {
language_servers: Default::default(),
},
);
- Ok(())
+
+ Ok(room)
}
pub fn unshare_project(
&mut self,
project_id: ProjectId,
connection_id: ConnectionId,
- ) -> Result<Project> {
+ ) -> Result<(&proto::Room, Project)> {
match self.projects.entry(project_id) {
btree_map::Entry::Occupied(e) => {
if e.get().host_connection_id == connection_id {
@@ -611,7 +647,20 @@ impl Store {
}
}
- Ok(project)
+ let room = self
+ .rooms
+ .get_mut(&project.room_id)
+ .ok_or_else(|| anyhow!("no such room"))?;
+ let participant = room
+ .participants
+ .iter_mut()
+ .find(|participant| participant.peer_id == connection_id.0)
+ .ok_or_else(|| anyhow!("no such room"))?;
+ participant
+ .project_ids
+ .retain(|id| *id != project_id.to_proto());
+
+ Ok((room, project))
} else {
Err(anyhow!("no such project"))?
}
@@ -731,8 +780,8 @@ impl Store {
pub fn leave_project(
&mut self,
- connection_id: ConnectionId,
project_id: ProjectId,
+ connection_id: ConnectionId,
) -> Result<LeftProject> {
let project = self
.projects
@@ -752,6 +801,7 @@ impl Store {
}
Ok(LeftProject {
+ id: project.id,
host_connection_id: project.host_connection_id,
host_user_id: project.host.user_id,
connection_ids: project.connection_ids(),
@@ -1049,13 +1049,13 @@ impl Project {
}
}
- pub fn share(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<u64>> {
+ pub fn share(&mut self, room_id: u64, cx: &mut ModelContext<Self>) -> Task<Result<u64>> {
if let ProjectClientState::Local { remote_id, .. } = &mut self.client_state {
if let Some(remote_id) = remote_id {
return Task::ready(Ok(*remote_id));
}
- let response = self.client.request(proto::ShareProject {});
+ let response = self.client.request(proto::ShareProject { room_id });
cx.spawn(|this, mut cx| async move {
let project_id = response.await?.project_id;
let mut worktree_share_tasks = Vec::new();
@@ -194,7 +194,9 @@ message RoomUpdated {
Room room = 1;
}
-message ShareProject {}
+message ShareProject {
+ uint64 room_id = 1;
+}
message ShareProjectResponse {
uint64 project_id = 1;