Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Wait for acknowledgment before sending the next project update

Browse source
This commit is contained in:
Antonio Scandurra 2022-11-14 15:32:49 +01:00
parent 65c5adff05
commit 40073f6100
11 changed files with 94 additions and 208 deletions
Download patch file Download diff file Expand all files Collapse all files

2
crates/call/src/room.rs
View file

@ -287,6 +287,8 @@ impl Room {
mut room: proto::Room,
cx: &mut ModelContext<Self>,
) -> Result<()> {
// TODO: honor room version.
// Filter ourselves out from the room's participants.
let local_participant_ix = room
.participants

16
crates/collab/src/db.rs
View file

@ -1145,8 +1145,8 @@ where
FROM projects, project_collaborators
WHERE
projects.room_id = $1 AND
projects.host_connection_id = $2 AND
projects.id = project_collaborators.project_id
projects.id = project_collaborators.project_id AND
project_collaborators.connection_id = $2
",
)
.bind(room_id)
@ -1370,9 +1370,9 @@ where
pub async fn share_project(
&self,
room_id: RoomId,
user_id: UserId,
connection_id: ConnectionId,
room_id: RoomId,
worktrees: &[proto::WorktreeMetadata],
) -> Result<(ProjectId, proto::Room)> {
test_support!(self, {
@ -1426,11 +1426,19 @@ where
.await?;
let room = self.commit_room_transaction(room_id, tx).await?;
dbg!(&room);
Ok((project_id, room))
})
}
// pub async fn join_project(
// &self,
// user_id: UserId,
// connection_id: ConnectionId,
// project_id: ProjectId,
// ) -> Result<(Project, ReplicaId)> {
// todo!()
// }
pub async fn unshare_project(&self, project_id: ProjectId) -> Result<()> {
todo!()
// test_support!(self, {

12
crates/collab/src/integration_tests.rs
View file

@ -30,9 +30,7 @@ use language::{
use live_kit_client::MacOSDisplay;
use lsp::{self, FakeLanguageServer};
use parking_lot::Mutex;
use project::{
search::SearchQuery, DiagnosticSummary, Project, ProjectPath, ProjectStore, WorktreeId,
};
use project::{search::SearchQuery, DiagnosticSummary, Project, ProjectPath, WorktreeId};
use rand::prelude::*;
use serde_json::json;
use settings::{Formatter, Settings};
@ -2280,7 +2278,6 @@ async fn test_leaving_project(
project_id,
client_b.client.clone(),
client_b.user_store.clone(),
client_b.project_store.clone(),
client_b.language_registry.clone(),
FakeFs::new(cx.background()),
cx,
@ -5792,11 +5789,9 @@ impl TestServer {
let fs = FakeFs::new(cx.background());
let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
let project_store = cx.add_model(|_| ProjectStore::new());
let app_state = Arc::new(workspace::AppState {
client: client.clone(),
user_store: user_store.clone(),
project_store: project_store.clone(),
languages: Arc::new(LanguageRegistry::new(Task::ready(()))),
themes: ThemeRegistry::new((), cx.font_cache()),
fs: fs.clone(),
@ -5823,7 +5818,6 @@ impl TestServer {
remote_projects: Default::default(),
next_root_dir_id: 0,
user_store,
project_store,
fs,
language_registry: Arc::new(LanguageRegistry::test()),
buffers: Default::default(),
@ -5929,7 +5923,6 @@ struct TestClient {
remote_projects: Vec<ModelHandle<Project>>,
next_root_dir_id: usize,
pub user_store: ModelHandle<UserStore>,
pub project_store: ModelHandle<ProjectStore>,
language_registry: Arc<LanguageRegistry>,
fs: Arc<FakeFs>,
buffers: HashMap<ModelHandle<Project>, HashSet<ModelHandle<language::Buffer>>>,
@ -5999,7 +5992,6 @@ impl TestClient {
Project::local(
self.client.clone(),
self.user_store.clone(),
self.project_store.clone(),
self.language_registry.clone(),
self.fs.clone(),
cx,
@ -6027,7 +6019,6 @@ impl TestClient {
host_project_id,
self.client.clone(),
self.user_store.clone(),
self.project_store.clone(),
self.language_registry.clone(),
FakeFs::new(cx.background()),
cx,
@ -6157,7 +6148,6 @@ impl TestClient {
remote_project_id,
client.client.clone(),
client.user_store.clone(),
client.project_store.clone(),
client.language_registry.clone(),
FakeFs::new(cx.background()),
cx.to_async(),

6
crates/collab/src/rpc.rs
View file

@ -151,7 +151,7 @@ impl Server {
.add_message_handler(Server::unshare_project)
.add_request_handler(Server::join_project)
.add_message_handler(Server::leave_project)
.add_message_handler(Server::update_project)
.add_request_handler(Server::update_project)
.add_request_handler(Server::update_worktree)
.add_message_handler(Server::start_language_server)
.add_message_handler(Server::update_language_server)
@ -861,9 +861,9 @@ impl Server {
.app_state
.db
.share_project(
RoomId::from_proto(request.payload.room_id),
request.sender_user_id,
request.sender_connection_id,
RoomId::from_proto(request.payload.room_id),
&request.payload.worktrees,
)
.await
@ -1084,6 +1084,7 @@ impl Server {
async fn update_project(
self: Arc<Server>,
request: Message<proto::UpdateProject>,
response: Response<proto::UpdateProject>,
) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id);
{
@ -1108,6 +1109,7 @@ impl Server {
},
);
self.room_updated(room);
response.send(proto::Ack {})?;
};
Ok(())

1
crates/collab_ui/src/collab_ui.rs
View file

@ -43,7 +43,6 @@ pub fn init(app_state: Arc<AppState>, cx: &mut MutableAppContext) {
project_id,
app_state.client.clone(),
app_state.user_store.clone(),
app_state.project_store.clone(),
app_state.languages.clone(),
app_state.fs.clone(),
cx.clone(),

218
crates/project/src/project.rs
View file

@ -70,10 +70,6 @@ pub trait Item: Entity {
fn entry_id(&self, cx: &AppContext) -> Option<ProjectEntryId>;
}
pub struct ProjectStore {
projects: Vec<WeakModelHandle<Project>>,
}
// Language server state is stored across 3 collections:
// language_servers =>
// a mapping from unique server id to LanguageServerState which can either be a task for a
@ -102,7 +98,6 @@ pub struct Project {
next_entry_id: Arc<AtomicUsize>,
next_diagnostic_group_id: usize,
user_store: ModelHandle<UserStore>,
project_store: ModelHandle<ProjectStore>,
fs: Arc<dyn Fs>,
client_state: Option<ProjectClientState>,
collaborators: HashMap<PeerId, Collaborator>,
@ -152,6 +147,8 @@ enum WorktreeHandle {
enum ProjectClientState {
Local {
remote_id: u64,
metadata_changed: watch::Sender<()>,
_maintain_metadata: Task<()>,
_detect_unshare: Task<Option<()>>,
},
Remote {
@ -376,7 +373,7 @@ impl Project {
client.add_model_message_handler(Self::handle_start_language_server);
client.add_model_message_handler(Self::handle_update_language_server);
client.add_model_message_handler(Self::handle_remove_collaborator);
client.add_model_message_handler(Self::handle_update_project);
client.add_model_message_handler(Self::handle_project_updated);
client.add_model_message_handler(Self::handle_unshare_project);
client.add_model_message_handler(Self::handle_create_buffer_for_peer);
client.add_model_message_handler(Self::handle_update_buffer_file);
@ -412,46 +409,39 @@ impl Project {
pub fn local(
client: Arc<Client>,
user_store: ModelHandle<UserStore>,
project_store: ModelHandle<ProjectStore>,
languages: Arc<LanguageRegistry>,
fs: Arc<dyn Fs>,
cx: &mut MutableAppContext,
) -> ModelHandle<Self> {
cx.add_model(|cx: &mut ModelContext<Self>| {
let handle = cx.weak_handle();
project_store.update(cx, |store, cx| store.add_project(handle, cx));
Self {
worktrees: Default::default(),
collaborators: Default::default(),
opened_buffers: Default::default(),
shared_buffers: Default::default(),
incomplete_buffers: Default::default(),
loading_buffers: Default::default(),
loading_local_worktrees: Default::default(),
buffer_snapshots: Default::default(),
client_state: None,
opened_buffer: watch::channel(),
client_subscriptions: Vec::new(),
_subscriptions: vec![cx.observe_global::<Settings, _>(Self::on_settings_changed)],
_maintain_buffer_languages: Self::maintain_buffer_languages(&languages, cx),
active_entry: None,
languages,
client,
user_store,
project_store,
fs,
next_entry_id: Default::default(),
next_diagnostic_group_id: Default::default(),
language_servers: Default::default(),
language_server_ids: Default::default(),
language_server_statuses: Default::default(),
last_workspace_edits_by_language_server: Default::default(),
language_server_settings: Default::default(),
buffers_being_formatted: Default::default(),
next_language_server_id: 0,
nonce: StdRng::from_entropy().gen(),
}
cx.add_model(|cx: &mut ModelContext<Self>| Self {
worktrees: Default::default(),
collaborators: Default::default(),
opened_buffers: Default::default(),
shared_buffers: Default::default(),
incomplete_buffers: Default::default(),
loading_buffers: Default::default(),
loading_local_worktrees: Default::default(),
buffer_snapshots: Default::default(),
client_state: None,
opened_buffer: watch::channel(),
client_subscriptions: Vec::new(),
_subscriptions: vec![cx.observe_global::<Settings, _>(Self::on_settings_changed)],
_maintain_buffer_languages: Self::maintain_buffer_languages(&languages, cx),
active_entry: None,
languages,
client,
user_store,
fs,
next_entry_id: Default::default(),
next_diagnostic_group_id: Default::default(),
language_servers: Default::default(),
language_server_ids: Default::default(),
language_server_statuses: Default::default(),
last_workspace_edits_by_language_server: Default::default(),
language_server_settings: Default::default(),
buffers_being_formatted: Default::default(),
next_language_server_id: 0,
nonce: StdRng::from_entropy().gen(),
})
}
@ -459,7 +449,6 @@ impl Project {
remote_id: u64,
client: Arc<Client>,
user_store: ModelHandle<UserStore>,
project_store: ModelHandle<ProjectStore>,
languages: Arc<LanguageRegistry>,
fs: Arc<dyn Fs>,
mut cx: AsyncAppContext,
@ -482,9 +471,6 @@ impl Project {
}
let this = cx.add_model(|cx: &mut ModelContext<Self>| {
let handle = cx.weak_handle();
project_store.update(cx, |store, cx| store.add_project(handle, cx));
let mut this = Self {
worktrees: Vec::new(),
loading_buffers: Default::default(),
@ -497,7 +483,6 @@ impl Project {
_maintain_buffer_languages: Self::maintain_buffer_languages(&languages, cx),
languages,
user_store: user_store.clone(),
project_store,
fs,
next_entry_id: Default::default(),
next_diagnostic_group_id: Default::default(),
@ -593,9 +578,7 @@ impl Project {
let http_client = client::test::FakeHttpClient::with_404_response();
let client = cx.update(|cx| client::Client::new(http_client.clone(), cx));
let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http_client, cx));
let project_store = cx.add_model(|_| ProjectStore::new());
let project =
cx.update(|cx| Project::local(client, user_store, project_store, languages, fs, cx));
let project = cx.update(|cx| Project::local(client, user_store, languages, fs, cx));
for path in root_paths {
let (tree, _) = project
.update(cx, |project, cx| {
@ -676,10 +659,6 @@ impl Project {
self.user_store.clone()
}
pub fn project_store(&self) -> ModelHandle<ProjectStore> {
self.project_store.clone()
}
#[cfg(any(test, feature = "test-support"))]
pub fn check_invariants(&self, cx: &AppContext) {
if self.is_local() {
@ -752,51 +731,12 @@ impl Project {
}
fn metadata_changed(&mut self, cx: &mut ModelContext<Self>) {
if let Some(ProjectClientState::Local { remote_id, .. }) = &self.client_state {
let project_id = *remote_id;
// Broadcast worktrees only if the project is online.
let worktrees = self
.worktrees
.iter()
.filter_map(|worktree| {
worktree
.upgrade(cx)
.map(|worktree| worktree.read(cx).as_local().unwrap().metadata_proto())
})
.collect();
self.client
.send(proto::UpdateProject {
project_id,
worktrees,
})
.log_err();
let worktrees = self.visible_worktrees(cx).collect::<Vec<_>>();
let scans_complete = futures::future::join_all(
worktrees
.iter()
.filter_map(|worktree| Some(worktree.read(cx).as_local()?.scan_complete())),
);
let worktrees = worktrees.into_iter().map(|handle| handle.downgrade());
cx.spawn_weak(move |_, cx| async move {
scans_complete.await;
cx.read(|cx| {
for worktree in worktrees {
if let Some(worktree) = worktree
.upgrade(cx)
.and_then(|worktree| worktree.read(cx).as_local())
{
worktree.send_extension_counts(project_id);
}
}
})
})
.detach();
if let Some(ProjectClientState::Local {
metadata_changed, ..
}) = &mut self.client_state
{
*metadata_changed.borrow_mut() = ();
}
self.project_store.update(cx, |_, cx| cx.notify());
cx.notify();
}
@ -1092,8 +1032,32 @@ impl Project {
cx.notify();
let mut status = self.client.status();
let (metadata_changed_tx, mut metadata_changed_rx) = watch::channel();
self.client_state = Some(ProjectClientState::Local {
remote_id: project_id,
metadata_changed: metadata_changed_tx,
_maintain_metadata: cx.spawn_weak(move |this, cx| async move {
while let Some(()) = metadata_changed_rx.next().await {
let Some(this) = this.upgrade(&cx) else { break };
this.read_with(&cx, |this, cx| {
let worktrees = this
.worktrees
.iter()
.filter_map(|worktree| {
worktree.upgrade(cx).map(|worktree| {
worktree.read(cx).as_local().unwrap().metadata_proto()
})
})
.collect();
this.client.request(proto::UpdateProject {
project_id,
worktrees,
})
})
.await
.log_err();
}
}),
_detect_unshare: cx.spawn_weak(move |this, mut cx| {
async move {
let is_connected = status.next().await.map_or(false, |s| s.is_connected());
@ -1632,10 +1596,6 @@ impl Project {
operations: vec![language::proto::serialize_operation(operation)],
});
cx.background().spawn(request).detach_and_log_err(cx);
} else if let Some(project_id) = self.remote_id() {
let _ = self
.client
.send(proto::RegisterProjectActivity { project_id });
}
}
BufferEvent::Edited { .. } => {
@ -4573,9 +4533,9 @@ impl Project {
})
}
async fn handle_update_project(
async fn handle_project_updated(
this: ModelHandle<Self>,
envelope: TypedEnvelope<proto::UpdateProject>,
envelope: TypedEnvelope<proto::ProjectUpdated>,
client: Arc<Client>,
mut cx: AsyncAppContext,
) -> Result<()> {
@ -5832,48 +5792,6 @@ impl Project {
}
}
impl ProjectStore {
pub fn new() -> Self {
Self {
projects: Default::default(),
}
}
pub fn projects<'a>(
&'a self,
cx: &'a AppContext,
) -> impl 'a + Iterator<Item = ModelHandle<Project>> {
self.projects
.iter()
.filter_map(|project| project.upgrade(cx))
}
fn add_project(&mut self, project: WeakModelHandle<Project>, cx: &mut ModelContext<Self>) {
if let Err(ix) = self
.projects
.binary_search_by_key(&project.id(), WeakModelHandle::id)
{
self.projects.insert(ix, project);
}
cx.notify();
}
fn prune_projects(&mut self, cx: &mut ModelContext<Self>) {
let mut did_change = false;
self.projects.retain(|project| {
if project.is_upgradable(cx) {
true
} else {
did_change = true;
false
}
});
if did_change {
cx.notify();
}
}
}
impl WorktreeHandle {
pub fn upgrade(&self, cx: &AppContext) -> Option<ModelHandle<Worktree>> {
match self {
@ -5952,16 +5870,10 @@ impl<'a> Iterator for PathMatchCandidateSetIter<'a> {
}
}
impl Entity for ProjectStore {
type Event = ();
}
impl Entity for Project {
type Event = Event;
fn release(&mut self, cx: &mut gpui::MutableAppContext) {
self.project_store.update(cx, ProjectStore::prune_projects);
fn release(&mut self, _: &mut gpui::MutableAppContext) {
match &self.client_state {
Some(ProjectClientState::Local { remote_id, .. }) => {
self.client

19
crates/project/src/worktree.rs
View file

@ -1051,25 +1051,6 @@ impl LocalWorktree {
pub fn is_shared(&self) -> bool {
self.share.is_some()
}
pub fn send_extension_counts(&self, project_id: u64) {
let mut extensions = Vec::new();
let mut counts = Vec::new();
for (extension, count) in self.extension_counts() {
extensions.push(extension.to_string_lossy().to_string());
counts.push(*count as u32);
}
self.client
.send(proto::UpdateWorktreeExtensions {
project_id,
worktree_id: self.id().to_proto(),
extensions,
counts,
})
.log_err();
}
}
impl RemoteWorktree {

7
crates/rpc/proto/zed.proto
View file

@ -48,9 +48,8 @@ message Envelope {
OpenBufferForSymbolResponse open_buffer_for_symbol_response = 40;
UpdateProject update_project = 41;
RegisterProjectActivity register_project_activity = 42;
ProjectUpdated project_updated = 42;
UpdateWorktree update_worktree = 43;
UpdateWorktreeExtensions update_worktree_extensions = 44;
CreateProjectEntry create_project_entry = 45;
RenameProjectEntry rename_project_entry = 46;
@ -258,8 +257,10 @@ message UpdateProject {
repeated WorktreeMetadata worktrees = 2;
}
message RegisterProjectActivity {
message ProjectUpdated {
uint64 project_id = 1;
repeated WorktreeMetadata worktrees = 2;
uint64 room_version = 3;
}
message JoinProject {

9
crates/rpc/src/proto.rs
View file

@ -140,12 +140,12 @@ messages!(
(OpenBufferResponse, Background),
(PerformRename, Background),
(PerformRenameResponse, Background),
(Ping, Foreground),
(PrepareRename, Background),
(PrepareRenameResponse, Background),
(ProjectEntryResponse, Foreground),
(ProjectUpdated, Foreground),
(RemoveContact, Foreground),
(Ping, Foreground),
(RegisterProjectActivity, Foreground),
(ReloadBuffers, Foreground),
(ReloadBuffersResponse, Foreground),
(RemoveProjectCollaborator, Foreground),
@ -175,7 +175,6 @@ messages!(
(UpdateParticipantLocation, Foreground),
(UpdateProject, Foreground),
(UpdateWorktree, Foreground),
(UpdateWorktreeExtensions, Background),
(UpdateDiffBase, Background),
(GetPrivateUserInfo, Foreground),
(GetPrivateUserInfoResponse, Foreground),
@ -231,6 +230,7 @@ request_messages!(
(Test, Test),
(UpdateBuffer, Ack),
(UpdateParticipantLocation, Ack),
(UpdateProject, Ack),
(UpdateWorktree, Ack),
);
@ -261,8 +261,8 @@ entity_messages!(
OpenBufferByPath,
OpenBufferForSymbol,
PerformRename,
ProjectUpdated,
PrepareRename,
RegisterProjectActivity,
ReloadBuffers,
RemoveProjectCollaborator,
RenameProjectEntry,
@ -278,7 +278,6 @@ entity_messages!(
UpdateLanguageServer,
UpdateProject,
UpdateWorktree,
UpdateWorktreeExtensions,
UpdateDiffBase
);

8
crates/workspace/src/workspace.rs
View file

@ -33,7 +33,7 @@ use log::{error, warn};
pub use pane::*;
pub use pane_group::*;
use postage::prelude::Stream;
use project::{Project, ProjectEntryId, ProjectPath, ProjectStore, Worktree, WorktreeId};
use project::{Project, ProjectEntryId, ProjectPath, Worktree, WorktreeId};
use searchable::SearchableItemHandle;
use serde::Deserialize;
use settings::{Autosave, DockAnchor, Settings};
@ -337,7 +337,6 @@ pub struct AppState {
pub themes: Arc<ThemeRegistry>,
pub client: Arc<client::Client>,
pub user_store: ModelHandle<client::UserStore>,
pub project_store: ModelHandle<ProjectStore>,
pub fs: Arc<dyn fs::Fs>,
pub build_window_options: fn() -> WindowOptions<'static>,
pub initialize_workspace: fn(&mut Workspace, &Arc<AppState>, &mut ViewContext<Workspace>),
@ -1039,7 +1038,6 @@ impl AppState {
let languages = Arc::new(LanguageRegistry::test());
let http_client = client::test::FakeHttpClient::with_404_response();
let client = Client::new(http_client.clone(), cx);
let project_store = cx.add_model(|_| ProjectStore::new());
let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http_client, cx));
let themes = ThemeRegistry::new((), cx.font_cache().clone());
Arc::new(Self {
@ -1048,7 +1046,6 @@ impl AppState {
fs,
languages,
user_store,
project_store,
initialize_workspace: |_, _, _| {},
build_window_options: Default::default,
default_item_factory: |_, _| unimplemented!(),
@ -1301,7 +1298,6 @@ impl Workspace {
Project::local(
app_state.client.clone(),
app_state.user_store.clone(),
app_state.project_store.clone(),
app_state.languages.clone(),
app_state.fs.clone(),
cx,
@ -2965,7 +2961,6 @@ pub fn open_paths(
let project = Project::local(
app_state.client.clone(),
app_state.user_store.clone(),
app_state.project_store.clone(),
app_state.languages.clone(),
app_state.fs.clone(),
cx,
@ -2997,7 +2992,6 @@ fn open_new(app_state: &Arc<AppState>, cx: &mut MutableAppContext) {
Project::local(
app_state.client.clone(),
app_state.user_store.clone(),
app_state.project_store.clone(),
app_state.languages.clone(),
app_state.fs.clone(),
cx,

4
crates/zed/src/main.rs
View file

@ -23,7 +23,7 @@ use isahc::{config::Configurable, Request};
use language::LanguageRegistry;
use log::LevelFilter;
use parking_lot::Mutex;
use project::{Fs, HomeDir, ProjectStore};
use project::{Fs, HomeDir};
use serde_json::json;
use settings::{
self, settings_file::SettingsFile, KeymapFileContent, Settings, SettingsFileContent,
@ -146,7 +146,6 @@ fn main() {
})
.detach();
let project_store = cx.add_model(|_| ProjectStore::new());
let db = cx.background().block(db);
client.start_telemetry(db.clone());
client.report_event("start app", Default::default());
@ -156,7 +155,6 @@ fn main() {
themes,
client: client.clone(),
user_store,
project_store,
fs,
build_window_options,
initialize_workspace,