1mod ignore;
2mod lsp_command;
3pub mod project_settings;
4pub mod search;
5pub mod terminals;
6pub mod worktree;
7
8#[cfg(test)]
9mod project_tests;
10#[cfg(test)]
11mod worktree_tests;
12
13use anyhow::{anyhow, Context, Result};
14use client::{proto, Client, TypedEnvelope, UserStore};
15use clock::ReplicaId;
16use collections::{hash_map, BTreeMap, HashMap, HashSet};
17use copilot::Copilot;
18use futures::{
19 channel::{
20 mpsc::{self, UnboundedReceiver},
21 oneshot,
22 },
23 future::{try_join_all, Shared},
24 stream::FuturesUnordered,
25 AsyncWriteExt, Future, FutureExt, StreamExt, TryFutureExt,
26};
27use globset::{Glob, GlobSet, GlobSetBuilder};
28use gpui::{
29 AnyModelHandle, AppContext, AsyncAppContext, BorrowAppContext, Entity, ModelContext,
30 ModelHandle, Task, WeakModelHandle,
31};
32use language::{
33 language_settings::{language_settings, FormatOnSave, Formatter},
34 point_to_lsp,
35 proto::{
36 deserialize_anchor, deserialize_fingerprint, deserialize_line_ending, deserialize_version,
37 serialize_anchor, serialize_version,
38 },
39 range_from_lsp, range_to_lsp, Anchor, Bias, Buffer, CachedLspAdapter, CodeAction, CodeLabel,
40 Completion, Diagnostic, DiagnosticEntry, DiagnosticSet, Diff, Event as BufferEvent, File as _,
41 Language, LanguageRegistry, LanguageServerName, LocalFile, LspAdapterDelegate, OffsetRangeExt,
42 Operation, Patch, PendingLanguageServer, PointUtf16, TextBufferSnapshot, ToOffset,
43 ToPointUtf16, Transaction, Unclipped,
44};
45use log::error;
46use lsp::{
47 DiagnosticSeverity, DiagnosticTag, DidChangeWatchedFilesRegistrationOptions,
48 DocumentHighlightKind, LanguageServer, LanguageServerId,
49};
50use lsp_command::*;
51use postage::watch;
52use project_settings::ProjectSettings;
53use rand::prelude::*;
54use search::SearchQuery;
55use serde::Serialize;
56use settings::SettingsStore;
57use sha2::{Digest, Sha256};
58use similar::{ChangeTag, TextDiff};
59use std::{
60 cell::RefCell,
61 cmp::{self, Ordering},
62 convert::TryInto,
63 hash::Hash,
64 mem,
65 num::NonZeroU32,
66 ops::Range,
67 path::{Component, Path, PathBuf},
68 rc::Rc,
69 str,
70 sync::{
71 atomic::{AtomicUsize, Ordering::SeqCst},
72 Arc,
73 },
74 time::{Duration, Instant},
75};
76use terminals::Terminals;
77use util::{
78 debug_panic, defer, http::HttpClient, merge_json_value_into,
79 paths::LOCAL_SETTINGS_RELATIVE_PATH, post_inc, ResultExt, TryFutureExt as _,
80};
81
82pub use fs::*;
83pub use worktree::*;
84
85pub trait Item {
86 fn entry_id(&self, cx: &AppContext) -> Option<ProjectEntryId>;
87 fn project_path(&self, cx: &AppContext) -> Option<ProjectPath>;
88}
89
90// Language server state is stored across 3 collections:
91// language_servers =>
92// a mapping from unique server id to LanguageServerState which can either be a task for a
93// server in the process of starting, or a running server with adapter and language server arcs
94// language_server_ids => a mapping from worktreeId and server name to the unique server id
95// language_server_statuses => a mapping from unique server id to the current server status
96//
97// Multiple worktrees can map to the same language server for example when you jump to the definition
98// of a file in the standard library. So language_server_ids is used to look up which server is active
99// for a given worktree and language server name
100//
101// When starting a language server, first the id map is checked to make sure a server isn't already available
102// for that worktree. If there is one, it finishes early. Otherwise, a new id is allocated and and
103// the Starting variant of LanguageServerState is stored in the language_servers map.
104pub struct Project {
105 worktrees: Vec<WorktreeHandle>,
106 active_entry: Option<ProjectEntryId>,
107 buffer_ordered_messages_tx: mpsc::UnboundedSender<BufferOrderedMessage>,
108 languages: Arc<LanguageRegistry>,
109 language_servers: HashMap<LanguageServerId, LanguageServerState>,
110 language_server_ids: HashMap<(WorktreeId, LanguageServerName), LanguageServerId>,
111 language_server_statuses: BTreeMap<LanguageServerId, LanguageServerStatus>,
112 last_workspace_edits_by_language_server: HashMap<LanguageServerId, ProjectTransaction>,
113 client: Arc<client::Client>,
114 next_entry_id: Arc<AtomicUsize>,
115 join_project_response_message_id: u32,
116 next_diagnostic_group_id: usize,
117 user_store: ModelHandle<UserStore>,
118 fs: Arc<dyn Fs>,
119 client_state: Option<ProjectClientState>,
120 collaborators: HashMap<proto::PeerId, Collaborator>,
121 client_subscriptions: Vec<client::Subscription>,
122 _subscriptions: Vec<gpui::Subscription>,
123 next_buffer_id: u64,
124 opened_buffer: (watch::Sender<()>, watch::Receiver<()>),
125 shared_buffers: HashMap<proto::PeerId, HashSet<u64>>,
126 #[allow(clippy::type_complexity)]
127 loading_buffers_by_path: HashMap<
128 ProjectPath,
129 postage::watch::Receiver<Option<Result<ModelHandle<Buffer>, Arc<anyhow::Error>>>>,
130 >,
131 #[allow(clippy::type_complexity)]
132 loading_local_worktrees:
133 HashMap<Arc<Path>, Shared<Task<Result<ModelHandle<Worktree>, Arc<anyhow::Error>>>>>,
134 opened_buffers: HashMap<u64, OpenBuffer>,
135 local_buffer_ids_by_path: HashMap<ProjectPath, u64>,
136 local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, u64>,
137 /// A mapping from a buffer ID to None means that we've started waiting for an ID but haven't finished loading it.
138 /// Used for re-issuing buffer requests when peers temporarily disconnect
139 incomplete_remote_buffers: HashMap<u64, Option<ModelHandle<Buffer>>>,
140 buffer_snapshots: HashMap<u64, HashMap<LanguageServerId, Vec<LspBufferSnapshot>>>, // buffer_id -> server_id -> vec of snapshots
141 buffers_being_formatted: HashSet<u64>,
142 buffers_needing_diff: HashSet<WeakModelHandle<Buffer>>,
143 git_diff_debouncer: DelayedDebounced,
144 nonce: u128,
145 _maintain_buffer_languages: Task<()>,
146 _maintain_workspace_config: Task<()>,
147 terminals: Terminals,
148 copilot_enabled: bool,
149}
150
151struct DelayedDebounced {
152 task: Option<Task<()>>,
153 cancel_channel: Option<oneshot::Sender<()>>,
154}
155
156impl DelayedDebounced {
157 fn new() -> DelayedDebounced {
158 DelayedDebounced {
159 task: None,
160 cancel_channel: None,
161 }
162 }
163
164 fn fire_new<F>(&mut self, delay: Duration, cx: &mut ModelContext<Project>, func: F)
165 where
166 F: 'static + FnOnce(&mut Project, &mut ModelContext<Project>) -> Task<()>,
167 {
168 if let Some(channel) = self.cancel_channel.take() {
169 _ = channel.send(());
170 }
171
172 let (sender, mut receiver) = oneshot::channel::<()>();
173 self.cancel_channel = Some(sender);
174
175 let previous_task = self.task.take();
176 self.task = Some(cx.spawn(|workspace, mut cx| async move {
177 let mut timer = cx.background().timer(delay).fuse();
178 if let Some(previous_task) = previous_task {
179 previous_task.await;
180 }
181
182 futures::select_biased! {
183 _ = receiver => return,
184 _ = timer => {}
185 }
186
187 workspace
188 .update(&mut cx, |workspace, cx| (func)(workspace, cx))
189 .await;
190 }));
191 }
192}
193
194struct LspBufferSnapshot {
195 version: i32,
196 snapshot: TextBufferSnapshot,
197}
198
199/// Message ordered with respect to buffer operations
200enum BufferOrderedMessage {
201 Operation {
202 buffer_id: u64,
203 operation: proto::Operation,
204 },
205 LanguageServerUpdate {
206 language_server_id: LanguageServerId,
207 message: proto::update_language_server::Variant,
208 },
209 Resync,
210}
211
212enum LocalProjectUpdate {
213 WorktreesChanged,
214 CreateBufferForPeer {
215 peer_id: proto::PeerId,
216 buffer_id: u64,
217 },
218}
219
220enum OpenBuffer {
221 Strong(ModelHandle<Buffer>),
222 Weak(WeakModelHandle<Buffer>),
223 Operations(Vec<Operation>),
224}
225
226enum WorktreeHandle {
227 Strong(ModelHandle<Worktree>),
228 Weak(WeakModelHandle<Worktree>),
229}
230
231enum ProjectClientState {
232 Local {
233 remote_id: u64,
234 updates_tx: mpsc::UnboundedSender<LocalProjectUpdate>,
235 _send_updates: Task<()>,
236 },
237 Remote {
238 sharing_has_stopped: bool,
239 remote_id: u64,
240 replica_id: ReplicaId,
241 },
242}
243
244#[derive(Clone, Debug)]
245pub struct Collaborator {
246 pub peer_id: proto::PeerId,
247 pub replica_id: ReplicaId,
248}
249
250#[derive(Clone, Debug, PartialEq)]
251pub enum Event {
252 LanguageServerAdded(LanguageServerId),
253 LanguageServerRemoved(LanguageServerId),
254 LanguageServerLog(LanguageServerId, String),
255 Notification(String),
256 ActiveEntryChanged(Option<ProjectEntryId>),
257 WorktreeAdded,
258 WorktreeRemoved(WorktreeId),
259 DiskBasedDiagnosticsStarted {
260 language_server_id: LanguageServerId,
261 },
262 DiskBasedDiagnosticsFinished {
263 language_server_id: LanguageServerId,
264 },
265 DiagnosticsUpdated {
266 path: ProjectPath,
267 language_server_id: LanguageServerId,
268 },
269 RemoteIdChanged(Option<u64>),
270 DisconnectedFromHost,
271 Closed,
272 DeletedEntry(ProjectEntryId),
273 CollaboratorUpdated {
274 old_peer_id: proto::PeerId,
275 new_peer_id: proto::PeerId,
276 },
277 CollaboratorLeft(proto::PeerId),
278}
279
280pub enum LanguageServerState {
281 Starting(Task<Option<Arc<LanguageServer>>>),
282 Running {
283 language: Arc<Language>,
284 adapter: Arc<CachedLspAdapter>,
285 server: Arc<LanguageServer>,
286 watched_paths: HashMap<WorktreeId, GlobSet>,
287 simulate_disk_based_diagnostics_completion: Option<Task<()>>,
288 },
289}
290
291#[derive(Serialize)]
292pub struct LanguageServerStatus {
293 pub name: String,
294 pub pending_work: BTreeMap<String, LanguageServerProgress>,
295 pub has_pending_diagnostic_updates: bool,
296 progress_tokens: HashSet<String>,
297}
298
299#[derive(Clone, Debug, Serialize)]
300pub struct LanguageServerProgress {
301 pub message: Option<String>,
302 pub percentage: Option<usize>,
303 #[serde(skip_serializing)]
304 pub last_update_at: Instant,
305}
306
307#[derive(Clone, Debug, Eq, PartialEq, Hash, PartialOrd, Ord)]
308pub struct ProjectPath {
309 pub worktree_id: WorktreeId,
310 pub path: Arc<Path>,
311}
312
313#[derive(Copy, Clone, Debug, Default, PartialEq, Serialize)]
314pub struct DiagnosticSummary {
315 pub error_count: usize,
316 pub warning_count: usize,
317}
318
319#[derive(Debug, Clone)]
320pub struct Location {
321 pub buffer: ModelHandle<Buffer>,
322 pub range: Range<language::Anchor>,
323}
324
325#[derive(Debug, Clone)]
326pub struct LocationLink {
327 pub origin: Option<Location>,
328 pub target: Location,
329}
330
331#[derive(Debug)]
332pub struct DocumentHighlight {
333 pub range: Range<language::Anchor>,
334 pub kind: DocumentHighlightKind,
335}
336
337#[derive(Clone, Debug)]
338pub struct Symbol {
339 pub language_server_name: LanguageServerName,
340 pub source_worktree_id: WorktreeId,
341 pub path: ProjectPath,
342 pub label: CodeLabel,
343 pub name: String,
344 pub kind: lsp::SymbolKind,
345 pub range: Range<Unclipped<PointUtf16>>,
346 pub signature: [u8; 32],
347}
348
349#[derive(Clone, Debug, PartialEq)]
350pub struct HoverBlock {
351 pub text: String,
352 pub kind: HoverBlockKind,
353}
354
355#[derive(Clone, Debug, PartialEq)]
356pub enum HoverBlockKind {
357 PlainText,
358 Markdown,
359 Code { language: String },
360}
361
362#[derive(Debug)]
363pub struct Hover {
364 pub contents: Vec<HoverBlock>,
365 pub range: Option<Range<language::Anchor>>,
366 pub language: Option<Arc<Language>>,
367}
368
369#[derive(Default)]
370pub struct ProjectTransaction(pub HashMap<ModelHandle<Buffer>, language::Transaction>);
371
372impl DiagnosticSummary {
373 fn new<'a, T: 'a>(diagnostics: impl IntoIterator<Item = &'a DiagnosticEntry<T>>) -> Self {
374 let mut this = Self {
375 error_count: 0,
376 warning_count: 0,
377 };
378
379 for entry in diagnostics {
380 if entry.diagnostic.is_primary {
381 match entry.diagnostic.severity {
382 DiagnosticSeverity::ERROR => this.error_count += 1,
383 DiagnosticSeverity::WARNING => this.warning_count += 1,
384 _ => {}
385 }
386 }
387 }
388
389 this
390 }
391
392 pub fn is_empty(&self) -> bool {
393 self.error_count == 0 && self.warning_count == 0
394 }
395
396 pub fn to_proto(
397 &self,
398 language_server_id: LanguageServerId,
399 path: &Path,
400 ) -> proto::DiagnosticSummary {
401 proto::DiagnosticSummary {
402 path: path.to_string_lossy().to_string(),
403 language_server_id: language_server_id.0 as u64,
404 error_count: self.error_count as u32,
405 warning_count: self.warning_count as u32,
406 }
407 }
408}
409
410#[derive(Clone, Copy, Debug, Default, Hash, PartialEq, Eq, PartialOrd, Ord)]
411pub struct ProjectEntryId(usize);
412
413impl ProjectEntryId {
414 pub const MAX: Self = Self(usize::MAX);
415
416 pub fn new(counter: &AtomicUsize) -> Self {
417 Self(counter.fetch_add(1, SeqCst))
418 }
419
420 pub fn from_proto(id: u64) -> Self {
421 Self(id as usize)
422 }
423
424 pub fn to_proto(&self) -> u64 {
425 self.0 as u64
426 }
427
428 pub fn to_usize(&self) -> usize {
429 self.0
430 }
431}
432
433#[derive(Debug, Clone, Copy, PartialEq, Eq)]
434pub enum FormatTrigger {
435 Save,
436 Manual,
437}
438
439struct ProjectLspAdapterDelegate {
440 project: ModelHandle<Project>,
441 http_client: Arc<dyn HttpClient>,
442}
443
444impl FormatTrigger {
445 fn from_proto(value: i32) -> FormatTrigger {
446 match value {
447 0 => FormatTrigger::Save,
448 1 => FormatTrigger::Manual,
449 _ => FormatTrigger::Save,
450 }
451 }
452}
453
454impl Project {
455 pub fn init_settings(cx: &mut AppContext) {
456 settings::register::<ProjectSettings>(cx);
457 }
458
459 pub fn init(client: &Arc<Client>, cx: &mut AppContext) {
460 Self::init_settings(cx);
461
462 client.add_model_message_handler(Self::handle_add_collaborator);
463 client.add_model_message_handler(Self::handle_update_project_collaborator);
464 client.add_model_message_handler(Self::handle_remove_collaborator);
465 client.add_model_message_handler(Self::handle_buffer_reloaded);
466 client.add_model_message_handler(Self::handle_buffer_saved);
467 client.add_model_message_handler(Self::handle_start_language_server);
468 client.add_model_message_handler(Self::handle_update_language_server);
469 client.add_model_message_handler(Self::handle_update_project);
470 client.add_model_message_handler(Self::handle_unshare_project);
471 client.add_model_message_handler(Self::handle_create_buffer_for_peer);
472 client.add_model_message_handler(Self::handle_update_buffer_file);
473 client.add_model_request_handler(Self::handle_update_buffer);
474 client.add_model_message_handler(Self::handle_update_diagnostic_summary);
475 client.add_model_message_handler(Self::handle_update_worktree);
476 client.add_model_message_handler(Self::handle_update_worktree_settings);
477 client.add_model_request_handler(Self::handle_create_project_entry);
478 client.add_model_request_handler(Self::handle_rename_project_entry);
479 client.add_model_request_handler(Self::handle_copy_project_entry);
480 client.add_model_request_handler(Self::handle_delete_project_entry);
481 client.add_model_request_handler(Self::handle_expand_project_entry);
482 client.add_model_request_handler(Self::handle_apply_additional_edits_for_completion);
483 client.add_model_request_handler(Self::handle_apply_code_action);
484 client.add_model_request_handler(Self::handle_on_type_formatting);
485 client.add_model_request_handler(Self::handle_reload_buffers);
486 client.add_model_request_handler(Self::handle_synchronize_buffers);
487 client.add_model_request_handler(Self::handle_format_buffers);
488 client.add_model_request_handler(Self::handle_lsp_command::<GetCodeActions>);
489 client.add_model_request_handler(Self::handle_lsp_command::<GetCompletions>);
490 client.add_model_request_handler(Self::handle_lsp_command::<GetHover>);
491 client.add_model_request_handler(Self::handle_lsp_command::<GetDefinition>);
492 client.add_model_request_handler(Self::handle_lsp_command::<GetTypeDefinition>);
493 client.add_model_request_handler(Self::handle_lsp_command::<GetDocumentHighlights>);
494 client.add_model_request_handler(Self::handle_lsp_command::<GetReferences>);
495 client.add_model_request_handler(Self::handle_lsp_command::<PrepareRename>);
496 client.add_model_request_handler(Self::handle_lsp_command::<PerformRename>);
497 client.add_model_request_handler(Self::handle_search_project);
498 client.add_model_request_handler(Self::handle_get_project_symbols);
499 client.add_model_request_handler(Self::handle_open_buffer_for_symbol);
500 client.add_model_request_handler(Self::handle_open_buffer_by_id);
501 client.add_model_request_handler(Self::handle_open_buffer_by_path);
502 client.add_model_request_handler(Self::handle_save_buffer);
503 client.add_model_message_handler(Self::handle_update_diff_base);
504 }
505
506 pub fn local(
507 client: Arc<Client>,
508 user_store: ModelHandle<UserStore>,
509 languages: Arc<LanguageRegistry>,
510 fs: Arc<dyn Fs>,
511 cx: &mut AppContext,
512 ) -> ModelHandle<Self> {
513 cx.add_model(|cx: &mut ModelContext<Self>| {
514 let (tx, rx) = mpsc::unbounded();
515 cx.spawn_weak(|this, cx| Self::send_buffer_ordered_messages(this, rx, cx))
516 .detach();
517 Self {
518 worktrees: Default::default(),
519 buffer_ordered_messages_tx: tx,
520 collaborators: Default::default(),
521 next_buffer_id: 0,
522 opened_buffers: Default::default(),
523 shared_buffers: Default::default(),
524 incomplete_remote_buffers: Default::default(),
525 loading_buffers_by_path: Default::default(),
526 loading_local_worktrees: Default::default(),
527 local_buffer_ids_by_path: Default::default(),
528 local_buffer_ids_by_entry_id: Default::default(),
529 buffer_snapshots: Default::default(),
530 join_project_response_message_id: 0,
531 client_state: None,
532 opened_buffer: watch::channel(),
533 client_subscriptions: Vec::new(),
534 _subscriptions: vec![
535 cx.observe_global::<SettingsStore, _>(Self::on_settings_changed)
536 ],
537 _maintain_buffer_languages: Self::maintain_buffer_languages(languages.clone(), cx),
538 _maintain_workspace_config: Self::maintain_workspace_config(languages.clone(), cx),
539 active_entry: None,
540 languages,
541 client,
542 user_store,
543 fs,
544 next_entry_id: Default::default(),
545 next_diagnostic_group_id: Default::default(),
546 language_servers: Default::default(),
547 language_server_ids: Default::default(),
548 language_server_statuses: Default::default(),
549 last_workspace_edits_by_language_server: Default::default(),
550 buffers_being_formatted: Default::default(),
551 buffers_needing_diff: Default::default(),
552 git_diff_debouncer: DelayedDebounced::new(),
553 nonce: StdRng::from_entropy().gen(),
554 terminals: Terminals {
555 local_handles: Vec::new(),
556 },
557 copilot_enabled: Copilot::global(cx).is_some(),
558 }
559 })
560 }
561
562 pub async fn remote(
563 remote_id: u64,
564 client: Arc<Client>,
565 user_store: ModelHandle<UserStore>,
566 languages: Arc<LanguageRegistry>,
567 fs: Arc<dyn Fs>,
568 mut cx: AsyncAppContext,
569 ) -> Result<ModelHandle<Self>> {
570 client.authenticate_and_connect(true, &cx).await?;
571
572 let subscription = client.subscribe_to_entity(remote_id)?;
573 let response = client
574 .request_envelope(proto::JoinProject {
575 project_id: remote_id,
576 })
577 .await?;
578 let this = cx.add_model(|cx| {
579 let replica_id = response.payload.replica_id as ReplicaId;
580
581 let mut worktrees = Vec::new();
582 for worktree in response.payload.worktrees {
583 let worktree = cx.update(|cx| {
584 Worktree::remote(remote_id, replica_id, worktree, client.clone(), cx)
585 });
586 worktrees.push(worktree);
587 }
588
589 let (tx, rx) = mpsc::unbounded();
590 cx.spawn_weak(|this, cx| Self::send_buffer_ordered_messages(this, rx, cx))
591 .detach();
592 let mut this = Self {
593 worktrees: Vec::new(),
594 buffer_ordered_messages_tx: tx,
595 loading_buffers_by_path: Default::default(),
596 next_buffer_id: 0,
597 opened_buffer: watch::channel(),
598 shared_buffers: Default::default(),
599 incomplete_remote_buffers: Default::default(),
600 loading_local_worktrees: Default::default(),
601 local_buffer_ids_by_path: Default::default(),
602 local_buffer_ids_by_entry_id: Default::default(),
603 active_entry: None,
604 collaborators: Default::default(),
605 join_project_response_message_id: response.message_id,
606 _maintain_buffer_languages: Self::maintain_buffer_languages(languages.clone(), cx),
607 _maintain_workspace_config: Self::maintain_workspace_config(languages.clone(), cx),
608 languages,
609 user_store: user_store.clone(),
610 fs,
611 next_entry_id: Default::default(),
612 next_diagnostic_group_id: Default::default(),
613 client_subscriptions: Default::default(),
614 _subscriptions: Default::default(),
615 client: client.clone(),
616 client_state: Some(ProjectClientState::Remote {
617 sharing_has_stopped: false,
618 remote_id,
619 replica_id,
620 }),
621 language_servers: Default::default(),
622 language_server_ids: Default::default(),
623 language_server_statuses: response
624 .payload
625 .language_servers
626 .into_iter()
627 .map(|server| {
628 (
629 LanguageServerId(server.id as usize),
630 LanguageServerStatus {
631 name: server.name,
632 pending_work: Default::default(),
633 has_pending_diagnostic_updates: false,
634 progress_tokens: Default::default(),
635 },
636 )
637 })
638 .collect(),
639 last_workspace_edits_by_language_server: Default::default(),
640 opened_buffers: Default::default(),
641 buffers_being_formatted: Default::default(),
642 buffers_needing_diff: Default::default(),
643 git_diff_debouncer: DelayedDebounced::new(),
644 buffer_snapshots: Default::default(),
645 nonce: StdRng::from_entropy().gen(),
646 terminals: Terminals {
647 local_handles: Vec::new(),
648 },
649 copilot_enabled: Copilot::global(cx).is_some(),
650 };
651 for worktree in worktrees {
652 let _ = this.add_worktree(&worktree, cx);
653 }
654 this
655 });
656 let subscription = subscription.set_model(&this, &mut cx);
657
658 let user_ids = response
659 .payload
660 .collaborators
661 .iter()
662 .map(|peer| peer.user_id)
663 .collect();
664 user_store
665 .update(&mut cx, |user_store, cx| user_store.get_users(user_ids, cx))
666 .await?;
667
668 this.update(&mut cx, |this, cx| {
669 this.set_collaborators_from_proto(response.payload.collaborators, cx)?;
670 this.client_subscriptions.push(subscription);
671 anyhow::Ok(())
672 })?;
673
674 Ok(this)
675 }
676
677 #[cfg(any(test, feature = "test-support"))]
678 pub async fn test(
679 fs: Arc<dyn Fs>,
680 root_paths: impl IntoIterator<Item = &Path>,
681 cx: &mut gpui::TestAppContext,
682 ) -> ModelHandle<Project> {
683 let mut languages = LanguageRegistry::test();
684 languages.set_executor(cx.background());
685 let http_client = util::http::FakeHttpClient::with_404_response();
686 let client = cx.update(|cx| client::Client::new(http_client.clone(), cx));
687 let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http_client, cx));
688 let project =
689 cx.update(|cx| Project::local(client, user_store, Arc::new(languages), fs, cx));
690 for path in root_paths {
691 let (tree, _) = project
692 .update(cx, |project, cx| {
693 project.find_or_create_local_worktree(path, true, cx)
694 })
695 .await
696 .unwrap();
697 tree.read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
698 .await;
699 }
700 project
701 }
702
703 fn on_settings_changed(&mut self, cx: &mut ModelContext<Self>) {
704 let mut language_servers_to_start = Vec::new();
705 for buffer in self.opened_buffers.values() {
706 if let Some(buffer) = buffer.upgrade(cx) {
707 let buffer = buffer.read(cx);
708 if let Some((file, language)) = buffer.file().zip(buffer.language()) {
709 let settings = language_settings(Some(language), Some(file), cx);
710 if settings.enable_language_server {
711 if let Some(file) = File::from_dyn(Some(file)) {
712 language_servers_to_start
713 .push((file.worktree.clone(), language.clone()));
714 }
715 }
716 }
717 }
718 }
719
720 let mut language_servers_to_stop = Vec::new();
721 let languages = self.languages.to_vec();
722 for (worktree_id, started_lsp_name) in self.language_server_ids.keys() {
723 let language = languages.iter().find(|l| {
724 l.lsp_adapters()
725 .iter()
726 .any(|adapter| &adapter.name == started_lsp_name)
727 });
728 if let Some(language) = language {
729 let worktree = self.worktree_for_id(*worktree_id, cx);
730 let file = worktree.and_then(|tree| {
731 tree.update(cx, |tree, cx| tree.root_file(cx).map(|f| f as _))
732 });
733 if !language_settings(Some(language), file.as_ref(), cx).enable_language_server {
734 language_servers_to_stop.push((*worktree_id, started_lsp_name.clone()));
735 }
736 }
737 }
738
739 // Stop all newly-disabled language servers.
740 for (worktree_id, adapter_name) in language_servers_to_stop {
741 self.stop_language_server(worktree_id, adapter_name, cx)
742 .detach();
743 }
744
745 // Start all the newly-enabled language servers.
746 for (worktree, language) in language_servers_to_start {
747 let worktree_path = worktree.read(cx).abs_path();
748 self.start_language_servers(&worktree, worktree_path, language, cx);
749 }
750
751 if !self.copilot_enabled && Copilot::global(cx).is_some() {
752 self.copilot_enabled = true;
753 for buffer in self.opened_buffers.values() {
754 if let Some(buffer) = buffer.upgrade(cx) {
755 self.register_buffer_with_copilot(&buffer, cx);
756 }
757 }
758 }
759
760 cx.notify();
761 }
762
763 pub fn buffer_for_id(&self, remote_id: u64, cx: &AppContext) -> Option<ModelHandle<Buffer>> {
764 self.opened_buffers
765 .get(&remote_id)
766 .and_then(|buffer| buffer.upgrade(cx))
767 }
768
769 pub fn languages(&self) -> &Arc<LanguageRegistry> {
770 &self.languages
771 }
772
773 pub fn client(&self) -> Arc<Client> {
774 self.client.clone()
775 }
776
777 pub fn user_store(&self) -> ModelHandle<UserStore> {
778 self.user_store.clone()
779 }
780
781 #[cfg(any(test, feature = "test-support"))]
782 pub fn opened_buffers(&self, cx: &AppContext) -> Vec<ModelHandle<Buffer>> {
783 self.opened_buffers
784 .values()
785 .filter_map(|b| b.upgrade(cx))
786 .collect()
787 }
788
789 #[cfg(any(test, feature = "test-support"))]
790 pub fn has_open_buffer(&self, path: impl Into<ProjectPath>, cx: &AppContext) -> bool {
791 let path = path.into();
792 if let Some(worktree) = self.worktree_for_id(path.worktree_id, cx) {
793 self.opened_buffers.iter().any(|(_, buffer)| {
794 if let Some(buffer) = buffer.upgrade(cx) {
795 if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
796 if file.worktree == worktree && file.path() == &path.path {
797 return true;
798 }
799 }
800 }
801 false
802 })
803 } else {
804 false
805 }
806 }
807
808 pub fn fs(&self) -> &Arc<dyn Fs> {
809 &self.fs
810 }
811
812 pub fn remote_id(&self) -> Option<u64> {
813 match self.client_state.as_ref()? {
814 ProjectClientState::Local { remote_id, .. }
815 | ProjectClientState::Remote { remote_id, .. } => Some(*remote_id),
816 }
817 }
818
819 pub fn replica_id(&self) -> ReplicaId {
820 match &self.client_state {
821 Some(ProjectClientState::Remote { replica_id, .. }) => *replica_id,
822 _ => 0,
823 }
824 }
825
826 fn metadata_changed(&mut self, cx: &mut ModelContext<Self>) {
827 if let Some(ProjectClientState::Local { updates_tx, .. }) = &mut self.client_state {
828 updates_tx
829 .unbounded_send(LocalProjectUpdate::WorktreesChanged)
830 .ok();
831 }
832 cx.notify();
833 }
834
835 pub fn collaborators(&self) -> &HashMap<proto::PeerId, Collaborator> {
836 &self.collaborators
837 }
838
839 /// Collect all worktrees, including ones that don't appear in the project panel
840 pub fn worktrees<'a>(
841 &'a self,
842 cx: &'a AppContext,
843 ) -> impl 'a + DoubleEndedIterator<Item = ModelHandle<Worktree>> {
844 self.worktrees
845 .iter()
846 .filter_map(move |worktree| worktree.upgrade(cx))
847 }
848
849 /// Collect all user-visible worktrees, the ones that appear in the project panel
850 pub fn visible_worktrees<'a>(
851 &'a self,
852 cx: &'a AppContext,
853 ) -> impl 'a + DoubleEndedIterator<Item = ModelHandle<Worktree>> {
854 self.worktrees.iter().filter_map(|worktree| {
855 worktree.upgrade(cx).and_then(|worktree| {
856 if worktree.read(cx).is_visible() {
857 Some(worktree)
858 } else {
859 None
860 }
861 })
862 })
863 }
864
865 pub fn worktree_root_names<'a>(&'a self, cx: &'a AppContext) -> impl Iterator<Item = &'a str> {
866 self.visible_worktrees(cx)
867 .map(|tree| tree.read(cx).root_name())
868 }
869
870 pub fn worktree_for_id(
871 &self,
872 id: WorktreeId,
873 cx: &AppContext,
874 ) -> Option<ModelHandle<Worktree>> {
875 self.worktrees(cx)
876 .find(|worktree| worktree.read(cx).id() == id)
877 }
878
879 pub fn worktree_for_entry(
880 &self,
881 entry_id: ProjectEntryId,
882 cx: &AppContext,
883 ) -> Option<ModelHandle<Worktree>> {
884 self.worktrees(cx)
885 .find(|worktree| worktree.read(cx).contains_entry(entry_id))
886 }
887
888 pub fn worktree_id_for_entry(
889 &self,
890 entry_id: ProjectEntryId,
891 cx: &AppContext,
892 ) -> Option<WorktreeId> {
893 self.worktree_for_entry(entry_id, cx)
894 .map(|worktree| worktree.read(cx).id())
895 }
896
897 pub fn contains_paths(&self, paths: &[PathBuf], cx: &AppContext) -> bool {
898 paths.iter().all(|path| self.contains_path(path, cx))
899 }
900
901 pub fn contains_path(&self, path: &Path, cx: &AppContext) -> bool {
902 for worktree in self.worktrees(cx) {
903 let worktree = worktree.read(cx).as_local();
904 if worktree.map_or(false, |w| w.contains_abs_path(path)) {
905 return true;
906 }
907 }
908 false
909 }
910
911 pub fn create_entry(
912 &mut self,
913 project_path: impl Into<ProjectPath>,
914 is_directory: bool,
915 cx: &mut ModelContext<Self>,
916 ) -> Option<Task<Result<Entry>>> {
917 let project_path = project_path.into();
918 let worktree = self.worktree_for_id(project_path.worktree_id, cx)?;
919 if self.is_local() {
920 Some(worktree.update(cx, |worktree, cx| {
921 worktree
922 .as_local_mut()
923 .unwrap()
924 .create_entry(project_path.path, is_directory, cx)
925 }))
926 } else {
927 let client = self.client.clone();
928 let project_id = self.remote_id().unwrap();
929 Some(cx.spawn_weak(|_, mut cx| async move {
930 let response = client
931 .request(proto::CreateProjectEntry {
932 worktree_id: project_path.worktree_id.to_proto(),
933 project_id,
934 path: project_path.path.to_string_lossy().into(),
935 is_directory,
936 })
937 .await?;
938 let entry = response
939 .entry
940 .ok_or_else(|| anyhow!("missing entry in response"))?;
941 worktree
942 .update(&mut cx, |worktree, cx| {
943 worktree.as_remote_mut().unwrap().insert_entry(
944 entry,
945 response.worktree_scan_id as usize,
946 cx,
947 )
948 })
949 .await
950 }))
951 }
952 }
953
954 pub fn copy_entry(
955 &mut self,
956 entry_id: ProjectEntryId,
957 new_path: impl Into<Arc<Path>>,
958 cx: &mut ModelContext<Self>,
959 ) -> Option<Task<Result<Entry>>> {
960 let worktree = self.worktree_for_entry(entry_id, cx)?;
961 let new_path = new_path.into();
962 if self.is_local() {
963 worktree.update(cx, |worktree, cx| {
964 worktree
965 .as_local_mut()
966 .unwrap()
967 .copy_entry(entry_id, new_path, cx)
968 })
969 } else {
970 let client = self.client.clone();
971 let project_id = self.remote_id().unwrap();
972
973 Some(cx.spawn_weak(|_, mut cx| async move {
974 let response = client
975 .request(proto::CopyProjectEntry {
976 project_id,
977 entry_id: entry_id.to_proto(),
978 new_path: new_path.to_string_lossy().into(),
979 })
980 .await?;
981 let entry = response
982 .entry
983 .ok_or_else(|| anyhow!("missing entry in response"))?;
984 worktree
985 .update(&mut cx, |worktree, cx| {
986 worktree.as_remote_mut().unwrap().insert_entry(
987 entry,
988 response.worktree_scan_id as usize,
989 cx,
990 )
991 })
992 .await
993 }))
994 }
995 }
996
997 pub fn rename_entry(
998 &mut self,
999 entry_id: ProjectEntryId,
1000 new_path: impl Into<Arc<Path>>,
1001 cx: &mut ModelContext<Self>,
1002 ) -> Option<Task<Result<Entry>>> {
1003 let worktree = self.worktree_for_entry(entry_id, cx)?;
1004 let new_path = new_path.into();
1005 if self.is_local() {
1006 worktree.update(cx, |worktree, cx| {
1007 worktree
1008 .as_local_mut()
1009 .unwrap()
1010 .rename_entry(entry_id, new_path, cx)
1011 })
1012 } else {
1013 let client = self.client.clone();
1014 let project_id = self.remote_id().unwrap();
1015
1016 Some(cx.spawn_weak(|_, mut cx| async move {
1017 let response = client
1018 .request(proto::RenameProjectEntry {
1019 project_id,
1020 entry_id: entry_id.to_proto(),
1021 new_path: new_path.to_string_lossy().into(),
1022 })
1023 .await?;
1024 let entry = response
1025 .entry
1026 .ok_or_else(|| anyhow!("missing entry in response"))?;
1027 worktree
1028 .update(&mut cx, |worktree, cx| {
1029 worktree.as_remote_mut().unwrap().insert_entry(
1030 entry,
1031 response.worktree_scan_id as usize,
1032 cx,
1033 )
1034 })
1035 .await
1036 }))
1037 }
1038 }
1039
1040 pub fn delete_entry(
1041 &mut self,
1042 entry_id: ProjectEntryId,
1043 cx: &mut ModelContext<Self>,
1044 ) -> Option<Task<Result<()>>> {
1045 let worktree = self.worktree_for_entry(entry_id, cx)?;
1046
1047 cx.emit(Event::DeletedEntry(entry_id));
1048
1049 if self.is_local() {
1050 worktree.update(cx, |worktree, cx| {
1051 worktree.as_local_mut().unwrap().delete_entry(entry_id, cx)
1052 })
1053 } else {
1054 let client = self.client.clone();
1055 let project_id = self.remote_id().unwrap();
1056 Some(cx.spawn_weak(|_, mut cx| async move {
1057 let response = client
1058 .request(proto::DeleteProjectEntry {
1059 project_id,
1060 entry_id: entry_id.to_proto(),
1061 })
1062 .await?;
1063 worktree
1064 .update(&mut cx, move |worktree, cx| {
1065 worktree.as_remote_mut().unwrap().delete_entry(
1066 entry_id,
1067 response.worktree_scan_id as usize,
1068 cx,
1069 )
1070 })
1071 .await
1072 }))
1073 }
1074 }
1075
1076 pub fn shared(&mut self, project_id: u64, cx: &mut ModelContext<Self>) -> Result<()> {
1077 if self.client_state.is_some() {
1078 return Err(anyhow!("project was already shared"));
1079 }
1080 self.client_subscriptions.push(
1081 self.client
1082 .subscribe_to_entity(project_id)?
1083 .set_model(&cx.handle(), &mut cx.to_async()),
1084 );
1085
1086 for open_buffer in self.opened_buffers.values_mut() {
1087 match open_buffer {
1088 OpenBuffer::Strong(_) => {}
1089 OpenBuffer::Weak(buffer) => {
1090 if let Some(buffer) = buffer.upgrade(cx) {
1091 *open_buffer = OpenBuffer::Strong(buffer);
1092 }
1093 }
1094 OpenBuffer::Operations(_) => unreachable!(),
1095 }
1096 }
1097
1098 for worktree_handle in self.worktrees.iter_mut() {
1099 match worktree_handle {
1100 WorktreeHandle::Strong(_) => {}
1101 WorktreeHandle::Weak(worktree) => {
1102 if let Some(worktree) = worktree.upgrade(cx) {
1103 *worktree_handle = WorktreeHandle::Strong(worktree);
1104 }
1105 }
1106 }
1107 }
1108
1109 for (server_id, status) in &self.language_server_statuses {
1110 self.client
1111 .send(proto::StartLanguageServer {
1112 project_id,
1113 server: Some(proto::LanguageServer {
1114 id: server_id.0 as u64,
1115 name: status.name.clone(),
1116 }),
1117 })
1118 .log_err();
1119 }
1120
1121 let store = cx.global::<SettingsStore>();
1122 for worktree in self.worktrees(cx) {
1123 let worktree_id = worktree.read(cx).id().to_proto();
1124 for (path, content) in store.local_settings(worktree.id()) {
1125 self.client
1126 .send(proto::UpdateWorktreeSettings {
1127 project_id,
1128 worktree_id,
1129 path: path.to_string_lossy().into(),
1130 content: Some(content),
1131 })
1132 .log_err();
1133 }
1134 }
1135
1136 let (updates_tx, mut updates_rx) = mpsc::unbounded();
1137 let client = self.client.clone();
1138 self.client_state = Some(ProjectClientState::Local {
1139 remote_id: project_id,
1140 updates_tx,
1141 _send_updates: cx.spawn_weak(move |this, mut cx| async move {
1142 while let Some(update) = updates_rx.next().await {
1143 let Some(this) = this.upgrade(&cx) else { break };
1144
1145 match update {
1146 LocalProjectUpdate::WorktreesChanged => {
1147 let worktrees = this
1148 .read_with(&cx, |this, cx| this.worktrees(cx).collect::<Vec<_>>());
1149 let update_project = this
1150 .read_with(&cx, |this, cx| {
1151 this.client.request(proto::UpdateProject {
1152 project_id,
1153 worktrees: this.worktree_metadata_protos(cx),
1154 })
1155 })
1156 .await;
1157 if update_project.is_ok() {
1158 for worktree in worktrees {
1159 worktree.update(&mut cx, |worktree, cx| {
1160 let worktree = worktree.as_local_mut().unwrap();
1161 worktree.share(project_id, cx).detach_and_log_err(cx)
1162 });
1163 }
1164 }
1165 }
1166 LocalProjectUpdate::CreateBufferForPeer { peer_id, buffer_id } => {
1167 let buffer = this.update(&mut cx, |this, _| {
1168 let buffer = this.opened_buffers.get(&buffer_id).unwrap();
1169 let shared_buffers =
1170 this.shared_buffers.entry(peer_id).or_default();
1171 if shared_buffers.insert(buffer_id) {
1172 if let OpenBuffer::Strong(buffer) = buffer {
1173 Some(buffer.clone())
1174 } else {
1175 None
1176 }
1177 } else {
1178 None
1179 }
1180 });
1181
1182 let Some(buffer) = buffer else { continue };
1183 let operations =
1184 buffer.read_with(&cx, |b, cx| b.serialize_ops(None, cx));
1185 let operations = operations.await;
1186 let state = buffer.read_with(&cx, |buffer, _| buffer.to_proto());
1187
1188 let initial_state = proto::CreateBufferForPeer {
1189 project_id,
1190 peer_id: Some(peer_id),
1191 variant: Some(proto::create_buffer_for_peer::Variant::State(state)),
1192 };
1193 if client.send(initial_state).log_err().is_some() {
1194 let client = client.clone();
1195 cx.background()
1196 .spawn(async move {
1197 let mut chunks = split_operations(operations).peekable();
1198 while let Some(chunk) = chunks.next() {
1199 let is_last = chunks.peek().is_none();
1200 client.send(proto::CreateBufferForPeer {
1201 project_id,
1202 peer_id: Some(peer_id),
1203 variant: Some(
1204 proto::create_buffer_for_peer::Variant::Chunk(
1205 proto::BufferChunk {
1206 buffer_id,
1207 operations: chunk,
1208 is_last,
1209 },
1210 ),
1211 ),
1212 })?;
1213 }
1214 anyhow::Ok(())
1215 })
1216 .await
1217 .log_err();
1218 }
1219 }
1220 }
1221 }
1222 }),
1223 });
1224
1225 self.metadata_changed(cx);
1226 cx.emit(Event::RemoteIdChanged(Some(project_id)));
1227 cx.notify();
1228 Ok(())
1229 }
1230
1231 pub fn reshared(
1232 &mut self,
1233 message: proto::ResharedProject,
1234 cx: &mut ModelContext<Self>,
1235 ) -> Result<()> {
1236 self.shared_buffers.clear();
1237 self.set_collaborators_from_proto(message.collaborators, cx)?;
1238 self.metadata_changed(cx);
1239 Ok(())
1240 }
1241
1242 pub fn rejoined(
1243 &mut self,
1244 message: proto::RejoinedProject,
1245 message_id: u32,
1246 cx: &mut ModelContext<Self>,
1247 ) -> Result<()> {
1248 cx.update_global::<SettingsStore, _, _>(|store, cx| {
1249 for worktree in &self.worktrees {
1250 store
1251 .clear_local_settings(worktree.handle_id(), cx)
1252 .log_err();
1253 }
1254 });
1255
1256 self.join_project_response_message_id = message_id;
1257 self.set_worktrees_from_proto(message.worktrees, cx)?;
1258 self.set_collaborators_from_proto(message.collaborators, cx)?;
1259 self.language_server_statuses = message
1260 .language_servers
1261 .into_iter()
1262 .map(|server| {
1263 (
1264 LanguageServerId(server.id as usize),
1265 LanguageServerStatus {
1266 name: server.name,
1267 pending_work: Default::default(),
1268 has_pending_diagnostic_updates: false,
1269 progress_tokens: Default::default(),
1270 },
1271 )
1272 })
1273 .collect();
1274 self.buffer_ordered_messages_tx
1275 .unbounded_send(BufferOrderedMessage::Resync)
1276 .unwrap();
1277 cx.notify();
1278 Ok(())
1279 }
1280
1281 pub fn unshare(&mut self, cx: &mut ModelContext<Self>) -> Result<()> {
1282 self.unshare_internal(cx)?;
1283 self.metadata_changed(cx);
1284 cx.notify();
1285 Ok(())
1286 }
1287
1288 fn unshare_internal(&mut self, cx: &mut AppContext) -> Result<()> {
1289 if self.is_remote() {
1290 return Err(anyhow!("attempted to unshare a remote project"));
1291 }
1292
1293 if let Some(ProjectClientState::Local { remote_id, .. }) = self.client_state.take() {
1294 self.collaborators.clear();
1295 self.shared_buffers.clear();
1296 self.client_subscriptions.clear();
1297
1298 for worktree_handle in self.worktrees.iter_mut() {
1299 if let WorktreeHandle::Strong(worktree) = worktree_handle {
1300 let is_visible = worktree.update(cx, |worktree, _| {
1301 worktree.as_local_mut().unwrap().unshare();
1302 worktree.is_visible()
1303 });
1304 if !is_visible {
1305 *worktree_handle = WorktreeHandle::Weak(worktree.downgrade());
1306 }
1307 }
1308 }
1309
1310 for open_buffer in self.opened_buffers.values_mut() {
1311 // Wake up any tasks waiting for peers' edits to this buffer.
1312 if let Some(buffer) = open_buffer.upgrade(cx) {
1313 buffer.update(cx, |buffer, _| buffer.give_up_waiting());
1314 }
1315
1316 if let OpenBuffer::Strong(buffer) = open_buffer {
1317 *open_buffer = OpenBuffer::Weak(buffer.downgrade());
1318 }
1319 }
1320
1321 self.client.send(proto::UnshareProject {
1322 project_id: remote_id,
1323 })?;
1324
1325 Ok(())
1326 } else {
1327 Err(anyhow!("attempted to unshare an unshared project"))
1328 }
1329 }
1330
1331 pub fn disconnected_from_host(&mut self, cx: &mut ModelContext<Self>) {
1332 self.disconnected_from_host_internal(cx);
1333 cx.emit(Event::DisconnectedFromHost);
1334 cx.notify();
1335 }
1336
1337 fn disconnected_from_host_internal(&mut self, cx: &mut AppContext) {
1338 if let Some(ProjectClientState::Remote {
1339 sharing_has_stopped,
1340 ..
1341 }) = &mut self.client_state
1342 {
1343 *sharing_has_stopped = true;
1344
1345 self.collaborators.clear();
1346
1347 for worktree in &self.worktrees {
1348 if let Some(worktree) = worktree.upgrade(cx) {
1349 worktree.update(cx, |worktree, _| {
1350 if let Some(worktree) = worktree.as_remote_mut() {
1351 worktree.disconnected_from_host();
1352 }
1353 });
1354 }
1355 }
1356
1357 for open_buffer in self.opened_buffers.values_mut() {
1358 // Wake up any tasks waiting for peers' edits to this buffer.
1359 if let Some(buffer) = open_buffer.upgrade(cx) {
1360 buffer.update(cx, |buffer, _| buffer.give_up_waiting());
1361 }
1362
1363 if let OpenBuffer::Strong(buffer) = open_buffer {
1364 *open_buffer = OpenBuffer::Weak(buffer.downgrade());
1365 }
1366 }
1367
1368 // Wake up all futures currently waiting on a buffer to get opened,
1369 // to give them a chance to fail now that we've disconnected.
1370 *self.opened_buffer.0.borrow_mut() = ();
1371 }
1372 }
1373
1374 pub fn close(&mut self, cx: &mut ModelContext<Self>) {
1375 cx.emit(Event::Closed);
1376 }
1377
1378 pub fn is_read_only(&self) -> bool {
1379 match &self.client_state {
1380 Some(ProjectClientState::Remote {
1381 sharing_has_stopped,
1382 ..
1383 }) => *sharing_has_stopped,
1384 _ => false,
1385 }
1386 }
1387
1388 pub fn is_local(&self) -> bool {
1389 match &self.client_state {
1390 Some(ProjectClientState::Remote { .. }) => false,
1391 _ => true,
1392 }
1393 }
1394
1395 pub fn is_remote(&self) -> bool {
1396 !self.is_local()
1397 }
1398
1399 pub fn create_buffer(
1400 &mut self,
1401 text: &str,
1402 language: Option<Arc<Language>>,
1403 cx: &mut ModelContext<Self>,
1404 ) -> Result<ModelHandle<Buffer>> {
1405 if self.is_remote() {
1406 return Err(anyhow!("creating buffers as a guest is not supported yet"));
1407 }
1408
1409 let buffer = cx.add_model(|cx| {
1410 Buffer::new(self.replica_id(), text, cx)
1411 .with_language(language.unwrap_or_else(|| language::PLAIN_TEXT.clone()), cx)
1412 });
1413 self.register_buffer(&buffer, cx)?;
1414 Ok(buffer)
1415 }
1416
1417 pub fn open_path(
1418 &mut self,
1419 path: impl Into<ProjectPath>,
1420 cx: &mut ModelContext<Self>,
1421 ) -> Task<Result<(ProjectEntryId, AnyModelHandle)>> {
1422 let task = self.open_buffer(path, cx);
1423 cx.spawn_weak(|_, cx| async move {
1424 let buffer = task.await?;
1425 let project_entry_id = buffer
1426 .read_with(&cx, |buffer, cx| {
1427 File::from_dyn(buffer.file()).and_then(|file| file.project_entry_id(cx))
1428 })
1429 .ok_or_else(|| anyhow!("no project entry"))?;
1430
1431 let buffer: &AnyModelHandle = &buffer;
1432 Ok((project_entry_id, buffer.clone()))
1433 })
1434 }
1435
1436 pub fn open_local_buffer(
1437 &mut self,
1438 abs_path: impl AsRef<Path>,
1439 cx: &mut ModelContext<Self>,
1440 ) -> Task<Result<ModelHandle<Buffer>>> {
1441 if let Some((worktree, relative_path)) = self.find_local_worktree(abs_path.as_ref(), cx) {
1442 self.open_buffer((worktree.read(cx).id(), relative_path), cx)
1443 } else {
1444 Task::ready(Err(anyhow!("no such path")))
1445 }
1446 }
1447
1448 pub fn open_buffer(
1449 &mut self,
1450 path: impl Into<ProjectPath>,
1451 cx: &mut ModelContext<Self>,
1452 ) -> Task<Result<ModelHandle<Buffer>>> {
1453 let project_path = path.into();
1454 let worktree = if let Some(worktree) = self.worktree_for_id(project_path.worktree_id, cx) {
1455 worktree
1456 } else {
1457 return Task::ready(Err(anyhow!("no such worktree")));
1458 };
1459
1460 // If there is already a buffer for the given path, then return it.
1461 let existing_buffer = self.get_open_buffer(&project_path, cx);
1462 if let Some(existing_buffer) = existing_buffer {
1463 return Task::ready(Ok(existing_buffer));
1464 }
1465
1466 let loading_watch = match self.loading_buffers_by_path.entry(project_path.clone()) {
1467 // If the given path is already being loaded, then wait for that existing
1468 // task to complete and return the same buffer.
1469 hash_map::Entry::Occupied(e) => e.get().clone(),
1470
1471 // Otherwise, record the fact that this path is now being loaded.
1472 hash_map::Entry::Vacant(entry) => {
1473 let (mut tx, rx) = postage::watch::channel();
1474 entry.insert(rx.clone());
1475
1476 let load_buffer = if worktree.read(cx).is_local() {
1477 self.open_local_buffer_internal(&project_path.path, &worktree, cx)
1478 } else {
1479 self.open_remote_buffer_internal(&project_path.path, &worktree, cx)
1480 };
1481
1482 cx.spawn(move |this, mut cx| async move {
1483 let load_result = load_buffer.await;
1484 *tx.borrow_mut() = Some(this.update(&mut cx, |this, _| {
1485 // Record the fact that the buffer is no longer loading.
1486 this.loading_buffers_by_path.remove(&project_path);
1487 let buffer = load_result.map_err(Arc::new)?;
1488 Ok(buffer)
1489 }));
1490 })
1491 .detach();
1492 rx
1493 }
1494 };
1495
1496 cx.foreground().spawn(async move {
1497 wait_for_loading_buffer(loading_watch)
1498 .await
1499 .map_err(|error| anyhow!("{}", error))
1500 })
1501 }
1502
1503 fn open_local_buffer_internal(
1504 &mut self,
1505 path: &Arc<Path>,
1506 worktree: &ModelHandle<Worktree>,
1507 cx: &mut ModelContext<Self>,
1508 ) -> Task<Result<ModelHandle<Buffer>>> {
1509 let buffer_id = post_inc(&mut self.next_buffer_id);
1510 let load_buffer = worktree.update(cx, |worktree, cx| {
1511 let worktree = worktree.as_local_mut().unwrap();
1512 worktree.load_buffer(buffer_id, path, cx)
1513 });
1514 cx.spawn(|this, mut cx| async move {
1515 let buffer = load_buffer.await?;
1516 this.update(&mut cx, |this, cx| this.register_buffer(&buffer, cx))?;
1517 Ok(buffer)
1518 })
1519 }
1520
1521 fn open_remote_buffer_internal(
1522 &mut self,
1523 path: &Arc<Path>,
1524 worktree: &ModelHandle<Worktree>,
1525 cx: &mut ModelContext<Self>,
1526 ) -> Task<Result<ModelHandle<Buffer>>> {
1527 let rpc = self.client.clone();
1528 let project_id = self.remote_id().unwrap();
1529 let remote_worktree_id = worktree.read(cx).id();
1530 let path = path.clone();
1531 let path_string = path.to_string_lossy().to_string();
1532 cx.spawn(|this, mut cx| async move {
1533 let response = rpc
1534 .request(proto::OpenBufferByPath {
1535 project_id,
1536 worktree_id: remote_worktree_id.to_proto(),
1537 path: path_string,
1538 })
1539 .await?;
1540 this.update(&mut cx, |this, cx| {
1541 this.wait_for_remote_buffer(response.buffer_id, cx)
1542 })
1543 .await
1544 })
1545 }
1546
1547 /// LanguageServerName is owned, because it is inserted into a map
1548 fn open_local_buffer_via_lsp(
1549 &mut self,
1550 abs_path: lsp::Url,
1551 language_server_id: LanguageServerId,
1552 language_server_name: LanguageServerName,
1553 cx: &mut ModelContext<Self>,
1554 ) -> Task<Result<ModelHandle<Buffer>>> {
1555 cx.spawn(|this, mut cx| async move {
1556 let abs_path = abs_path
1557 .to_file_path()
1558 .map_err(|_| anyhow!("can't convert URI to path"))?;
1559 let (worktree, relative_path) = if let Some(result) =
1560 this.read_with(&cx, |this, cx| this.find_local_worktree(&abs_path, cx))
1561 {
1562 result
1563 } else {
1564 let worktree = this
1565 .update(&mut cx, |this, cx| {
1566 this.create_local_worktree(&abs_path, false, cx)
1567 })
1568 .await?;
1569 this.update(&mut cx, |this, cx| {
1570 this.language_server_ids.insert(
1571 (worktree.read(cx).id(), language_server_name),
1572 language_server_id,
1573 );
1574 });
1575 (worktree, PathBuf::new())
1576 };
1577
1578 let project_path = ProjectPath {
1579 worktree_id: worktree.read_with(&cx, |worktree, _| worktree.id()),
1580 path: relative_path.into(),
1581 };
1582 this.update(&mut cx, |this, cx| this.open_buffer(project_path, cx))
1583 .await
1584 })
1585 }
1586
1587 pub fn open_buffer_by_id(
1588 &mut self,
1589 id: u64,
1590 cx: &mut ModelContext<Self>,
1591 ) -> Task<Result<ModelHandle<Buffer>>> {
1592 if let Some(buffer) = self.buffer_for_id(id, cx) {
1593 Task::ready(Ok(buffer))
1594 } else if self.is_local() {
1595 Task::ready(Err(anyhow!("buffer {} does not exist", id)))
1596 } else if let Some(project_id) = self.remote_id() {
1597 let request = self
1598 .client
1599 .request(proto::OpenBufferById { project_id, id });
1600 cx.spawn(|this, mut cx| async move {
1601 let buffer_id = request.await?.buffer_id;
1602 this.update(&mut cx, |this, cx| {
1603 this.wait_for_remote_buffer(buffer_id, cx)
1604 })
1605 .await
1606 })
1607 } else {
1608 Task::ready(Err(anyhow!("cannot open buffer while disconnected")))
1609 }
1610 }
1611
1612 pub fn save_buffers(
1613 &self,
1614 buffers: HashSet<ModelHandle<Buffer>>,
1615 cx: &mut ModelContext<Self>,
1616 ) -> Task<Result<()>> {
1617 cx.spawn(|this, mut cx| async move {
1618 let save_tasks = buffers
1619 .into_iter()
1620 .map(|buffer| this.update(&mut cx, |this, cx| this.save_buffer(buffer, cx)));
1621 try_join_all(save_tasks).await?;
1622 Ok(())
1623 })
1624 }
1625
1626 pub fn save_buffer(
1627 &self,
1628 buffer: ModelHandle<Buffer>,
1629 cx: &mut ModelContext<Self>,
1630 ) -> Task<Result<()>> {
1631 let Some(file) = File::from_dyn(buffer.read(cx).file()) else {
1632 return Task::ready(Err(anyhow!("buffer doesn't have a file")));
1633 };
1634 let worktree = file.worktree.clone();
1635 let path = file.path.clone();
1636 worktree.update(cx, |worktree, cx| match worktree {
1637 Worktree::Local(worktree) => worktree.save_buffer(buffer, path, false, cx),
1638 Worktree::Remote(worktree) => worktree.save_buffer(buffer, cx),
1639 })
1640 }
1641
1642 pub fn save_buffer_as(
1643 &mut self,
1644 buffer: ModelHandle<Buffer>,
1645 abs_path: PathBuf,
1646 cx: &mut ModelContext<Self>,
1647 ) -> Task<Result<()>> {
1648 let worktree_task = self.find_or_create_local_worktree(&abs_path, true, cx);
1649 let old_file = File::from_dyn(buffer.read(cx).file())
1650 .filter(|f| f.is_local())
1651 .cloned();
1652 cx.spawn(|this, mut cx| async move {
1653 if let Some(old_file) = &old_file {
1654 this.update(&mut cx, |this, cx| {
1655 this.unregister_buffer_from_language_servers(&buffer, old_file, cx);
1656 });
1657 }
1658 let (worktree, path) = worktree_task.await?;
1659 worktree
1660 .update(&mut cx, |worktree, cx| match worktree {
1661 Worktree::Local(worktree) => {
1662 worktree.save_buffer(buffer.clone(), path.into(), true, cx)
1663 }
1664 Worktree::Remote(_) => panic!("cannot remote buffers as new files"),
1665 })
1666 .await?;
1667 this.update(&mut cx, |this, cx| {
1668 this.detect_language_for_buffer(&buffer, cx);
1669 this.register_buffer_with_language_servers(&buffer, cx);
1670 });
1671 Ok(())
1672 })
1673 }
1674
1675 pub fn get_open_buffer(
1676 &mut self,
1677 path: &ProjectPath,
1678 cx: &mut ModelContext<Self>,
1679 ) -> Option<ModelHandle<Buffer>> {
1680 let worktree = self.worktree_for_id(path.worktree_id, cx)?;
1681 self.opened_buffers.values().find_map(|buffer| {
1682 let buffer = buffer.upgrade(cx)?;
1683 let file = File::from_dyn(buffer.read(cx).file())?;
1684 if file.worktree == worktree && file.path() == &path.path {
1685 Some(buffer)
1686 } else {
1687 None
1688 }
1689 })
1690 }
1691
1692 fn register_buffer(
1693 &mut self,
1694 buffer: &ModelHandle<Buffer>,
1695 cx: &mut ModelContext<Self>,
1696 ) -> Result<()> {
1697 self.request_buffer_diff_recalculation(buffer, cx);
1698 buffer.update(cx, |buffer, _| {
1699 buffer.set_language_registry(self.languages.clone())
1700 });
1701
1702 let remote_id = buffer.read(cx).remote_id();
1703 let is_remote = self.is_remote();
1704 let open_buffer = if is_remote || self.is_shared() {
1705 OpenBuffer::Strong(buffer.clone())
1706 } else {
1707 OpenBuffer::Weak(buffer.downgrade())
1708 };
1709
1710 match self.opened_buffers.entry(remote_id) {
1711 hash_map::Entry::Vacant(entry) => {
1712 entry.insert(open_buffer);
1713 }
1714 hash_map::Entry::Occupied(mut entry) => {
1715 if let OpenBuffer::Operations(operations) = entry.get_mut() {
1716 buffer.update(cx, |b, cx| b.apply_ops(operations.drain(..), cx))?;
1717 } else if entry.get().upgrade(cx).is_some() {
1718 if is_remote {
1719 return Ok(());
1720 } else {
1721 debug_panic!("buffer {} was already registered", remote_id);
1722 Err(anyhow!("buffer {} was already registered", remote_id))?;
1723 }
1724 }
1725 entry.insert(open_buffer);
1726 }
1727 }
1728 cx.subscribe(buffer, |this, buffer, event, cx| {
1729 this.on_buffer_event(buffer, event, cx);
1730 })
1731 .detach();
1732
1733 if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
1734 if file.is_local {
1735 self.local_buffer_ids_by_path.insert(
1736 ProjectPath {
1737 worktree_id: file.worktree_id(cx),
1738 path: file.path.clone(),
1739 },
1740 remote_id,
1741 );
1742
1743 self.local_buffer_ids_by_entry_id
1744 .insert(file.entry_id, remote_id);
1745 }
1746 }
1747
1748 self.detect_language_for_buffer(buffer, cx);
1749 self.register_buffer_with_language_servers(buffer, cx);
1750 self.register_buffer_with_copilot(buffer, cx);
1751 cx.observe_release(buffer, |this, buffer, cx| {
1752 if let Some(file) = File::from_dyn(buffer.file()) {
1753 if file.is_local() {
1754 let uri = lsp::Url::from_file_path(file.abs_path(cx)).unwrap();
1755 for server in this.language_servers_for_buffer(buffer, cx) {
1756 server
1757 .1
1758 .notify::<lsp::notification::DidCloseTextDocument>(
1759 lsp::DidCloseTextDocumentParams {
1760 text_document: lsp::TextDocumentIdentifier::new(uri.clone()),
1761 },
1762 )
1763 .log_err();
1764 }
1765 }
1766 }
1767 })
1768 .detach();
1769
1770 *self.opened_buffer.0.borrow_mut() = ();
1771 Ok(())
1772 }
1773
1774 fn register_buffer_with_language_servers(
1775 &mut self,
1776 buffer_handle: &ModelHandle<Buffer>,
1777 cx: &mut ModelContext<Self>,
1778 ) {
1779 let buffer = buffer_handle.read(cx);
1780 let buffer_id = buffer.remote_id();
1781
1782 if let Some(file) = File::from_dyn(buffer.file()) {
1783 if !file.is_local() {
1784 return;
1785 }
1786
1787 let uri = lsp::Url::from_file_path(file.abs_path(cx)).unwrap();
1788 let initial_snapshot = buffer.text_snapshot();
1789 let language = buffer.language().cloned();
1790 let worktree_id = file.worktree_id(cx);
1791
1792 if let Some(local_worktree) = file.worktree.read(cx).as_local() {
1793 for (server_id, diagnostics) in local_worktree.diagnostics_for_path(file.path()) {
1794 self.update_buffer_diagnostics(buffer_handle, server_id, None, diagnostics, cx)
1795 .log_err();
1796 }
1797 }
1798
1799 if let Some(language) = language {
1800 for adapter in language.lsp_adapters() {
1801 let language_id = adapter.language_ids.get(language.name().as_ref()).cloned();
1802 let server = self
1803 .language_server_ids
1804 .get(&(worktree_id, adapter.name.clone()))
1805 .and_then(|id| self.language_servers.get(id))
1806 .and_then(|server_state| {
1807 if let LanguageServerState::Running { server, .. } = server_state {
1808 Some(server.clone())
1809 } else {
1810 None
1811 }
1812 });
1813 let server = match server {
1814 Some(server) => server,
1815 None => continue,
1816 };
1817
1818 server
1819 .notify::<lsp::notification::DidOpenTextDocument>(
1820 lsp::DidOpenTextDocumentParams {
1821 text_document: lsp::TextDocumentItem::new(
1822 uri.clone(),
1823 language_id.unwrap_or_default(),
1824 0,
1825 initial_snapshot.text(),
1826 ),
1827 },
1828 )
1829 .log_err();
1830
1831 buffer_handle.update(cx, |buffer, cx| {
1832 buffer.set_completion_triggers(
1833 server
1834 .capabilities()
1835 .completion_provider
1836 .as_ref()
1837 .and_then(|provider| provider.trigger_characters.clone())
1838 .unwrap_or_default(),
1839 cx,
1840 );
1841 });
1842
1843 let snapshot = LspBufferSnapshot {
1844 version: 0,
1845 snapshot: initial_snapshot.clone(),
1846 };
1847 self.buffer_snapshots
1848 .entry(buffer_id)
1849 .or_default()
1850 .insert(server.server_id(), vec![snapshot]);
1851 }
1852 }
1853 }
1854 }
1855
1856 fn unregister_buffer_from_language_servers(
1857 &mut self,
1858 buffer: &ModelHandle<Buffer>,
1859 old_file: &File,
1860 cx: &mut ModelContext<Self>,
1861 ) {
1862 let old_path = match old_file.as_local() {
1863 Some(local) => local.abs_path(cx),
1864 None => return,
1865 };
1866
1867 buffer.update(cx, |buffer, cx| {
1868 let worktree_id = old_file.worktree_id(cx);
1869 let ids = &self.language_server_ids;
1870
1871 let language = buffer.language().cloned();
1872 let adapters = language.iter().flat_map(|language| language.lsp_adapters());
1873 for &server_id in adapters.flat_map(|a| ids.get(&(worktree_id, a.name.clone()))) {
1874 buffer.update_diagnostics(server_id, Default::default(), cx);
1875 }
1876
1877 self.buffer_snapshots.remove(&buffer.remote_id());
1878 let file_url = lsp::Url::from_file_path(old_path).unwrap();
1879 for (_, language_server) in self.language_servers_for_buffer(buffer, cx) {
1880 language_server
1881 .notify::<lsp::notification::DidCloseTextDocument>(
1882 lsp::DidCloseTextDocumentParams {
1883 text_document: lsp::TextDocumentIdentifier::new(file_url.clone()),
1884 },
1885 )
1886 .log_err();
1887 }
1888 });
1889 }
1890
1891 fn register_buffer_with_copilot(
1892 &self,
1893 buffer_handle: &ModelHandle<Buffer>,
1894 cx: &mut ModelContext<Self>,
1895 ) {
1896 if let Some(copilot) = Copilot::global(cx) {
1897 copilot.update(cx, |copilot, cx| copilot.register_buffer(buffer_handle, cx));
1898 }
1899 }
1900
1901 async fn send_buffer_ordered_messages(
1902 this: WeakModelHandle<Self>,
1903 rx: UnboundedReceiver<BufferOrderedMessage>,
1904 mut cx: AsyncAppContext,
1905 ) -> Option<()> {
1906 const MAX_BATCH_SIZE: usize = 128;
1907
1908 let mut operations_by_buffer_id = HashMap::default();
1909 async fn flush_operations(
1910 this: &ModelHandle<Project>,
1911 operations_by_buffer_id: &mut HashMap<u64, Vec<proto::Operation>>,
1912 needs_resync_with_host: &mut bool,
1913 is_local: bool,
1914 cx: &AsyncAppContext,
1915 ) {
1916 for (buffer_id, operations) in operations_by_buffer_id.drain() {
1917 let request = this.read_with(cx, |this, _| {
1918 let project_id = this.remote_id()?;
1919 Some(this.client.request(proto::UpdateBuffer {
1920 buffer_id,
1921 project_id,
1922 operations,
1923 }))
1924 });
1925 if let Some(request) = request {
1926 if request.await.is_err() && !is_local {
1927 *needs_resync_with_host = true;
1928 break;
1929 }
1930 }
1931 }
1932 }
1933
1934 let mut needs_resync_with_host = false;
1935 let mut changes = rx.ready_chunks(MAX_BATCH_SIZE);
1936
1937 while let Some(changes) = changes.next().await {
1938 let this = this.upgrade(&mut cx)?;
1939 let is_local = this.read_with(&cx, |this, _| this.is_local());
1940
1941 for change in changes {
1942 match change {
1943 BufferOrderedMessage::Operation {
1944 buffer_id,
1945 operation,
1946 } => {
1947 if needs_resync_with_host {
1948 continue;
1949 }
1950
1951 operations_by_buffer_id
1952 .entry(buffer_id)
1953 .or_insert(Vec::new())
1954 .push(operation);
1955 }
1956
1957 BufferOrderedMessage::Resync => {
1958 operations_by_buffer_id.clear();
1959 if this
1960 .update(&mut cx, |this, cx| this.synchronize_remote_buffers(cx))
1961 .await
1962 .is_ok()
1963 {
1964 needs_resync_with_host = false;
1965 }
1966 }
1967
1968 BufferOrderedMessage::LanguageServerUpdate {
1969 language_server_id,
1970 message,
1971 } => {
1972 flush_operations(
1973 &this,
1974 &mut operations_by_buffer_id,
1975 &mut needs_resync_with_host,
1976 is_local,
1977 &cx,
1978 )
1979 .await;
1980
1981 this.read_with(&cx, |this, _| {
1982 if let Some(project_id) = this.remote_id() {
1983 this.client
1984 .send(proto::UpdateLanguageServer {
1985 project_id,
1986 language_server_id: language_server_id.0 as u64,
1987 variant: Some(message),
1988 })
1989 .log_err();
1990 }
1991 });
1992 }
1993 }
1994 }
1995
1996 flush_operations(
1997 &this,
1998 &mut operations_by_buffer_id,
1999 &mut needs_resync_with_host,
2000 is_local,
2001 &cx,
2002 )
2003 .await;
2004 }
2005
2006 None
2007 }
2008
2009 fn on_buffer_event(
2010 &mut self,
2011 buffer: ModelHandle<Buffer>,
2012 event: &BufferEvent,
2013 cx: &mut ModelContext<Self>,
2014 ) -> Option<()> {
2015 if matches!(
2016 event,
2017 BufferEvent::Edited { .. } | BufferEvent::Reloaded | BufferEvent::DiffBaseChanged
2018 ) {
2019 self.request_buffer_diff_recalculation(&buffer, cx);
2020 }
2021
2022 match event {
2023 BufferEvent::Operation(operation) => {
2024 self.buffer_ordered_messages_tx
2025 .unbounded_send(BufferOrderedMessage::Operation {
2026 buffer_id: buffer.read(cx).remote_id(),
2027 operation: language::proto::serialize_operation(operation),
2028 })
2029 .ok();
2030 }
2031
2032 BufferEvent::Edited { .. } => {
2033 let buffer = buffer.read(cx);
2034 let file = File::from_dyn(buffer.file())?;
2035 let abs_path = file.as_local()?.abs_path(cx);
2036 let uri = lsp::Url::from_file_path(abs_path).unwrap();
2037 let next_snapshot = buffer.text_snapshot();
2038
2039 let language_servers: Vec<_> = self
2040 .language_servers_for_buffer(buffer, cx)
2041 .map(|i| i.1.clone())
2042 .collect();
2043
2044 for language_server in language_servers {
2045 let language_server = language_server.clone();
2046
2047 let buffer_snapshots = self
2048 .buffer_snapshots
2049 .get_mut(&buffer.remote_id())
2050 .and_then(|m| m.get_mut(&language_server.server_id()))?;
2051 let previous_snapshot = buffer_snapshots.last()?;
2052 let next_version = previous_snapshot.version + 1;
2053
2054 let content_changes = buffer
2055 .edits_since::<(PointUtf16, usize)>(previous_snapshot.snapshot.version())
2056 .map(|edit| {
2057 let edit_start = edit.new.start.0;
2058 let edit_end = edit_start + (edit.old.end.0 - edit.old.start.0);
2059 let new_text = next_snapshot
2060 .text_for_range(edit.new.start.1..edit.new.end.1)
2061 .collect();
2062 lsp::TextDocumentContentChangeEvent {
2063 range: Some(lsp::Range::new(
2064 point_to_lsp(edit_start),
2065 point_to_lsp(edit_end),
2066 )),
2067 range_length: None,
2068 text: new_text,
2069 }
2070 })
2071 .collect();
2072
2073 buffer_snapshots.push(LspBufferSnapshot {
2074 version: next_version,
2075 snapshot: next_snapshot.clone(),
2076 });
2077
2078 language_server
2079 .notify::<lsp::notification::DidChangeTextDocument>(
2080 lsp::DidChangeTextDocumentParams {
2081 text_document: lsp::VersionedTextDocumentIdentifier::new(
2082 uri.clone(),
2083 next_version,
2084 ),
2085 content_changes,
2086 },
2087 )
2088 .log_err();
2089 }
2090 }
2091
2092 BufferEvent::Saved => {
2093 let file = File::from_dyn(buffer.read(cx).file())?;
2094 let worktree_id = file.worktree_id(cx);
2095 let abs_path = file.as_local()?.abs_path(cx);
2096 let text_document = lsp::TextDocumentIdentifier {
2097 uri: lsp::Url::from_file_path(abs_path).unwrap(),
2098 };
2099
2100 for (_, _, server) in self.language_servers_for_worktree(worktree_id) {
2101 server
2102 .notify::<lsp::notification::DidSaveTextDocument>(
2103 lsp::DidSaveTextDocumentParams {
2104 text_document: text_document.clone(),
2105 text: None,
2106 },
2107 )
2108 .log_err();
2109 }
2110
2111 let language_server_ids = self.language_server_ids_for_buffer(buffer.read(cx), cx);
2112 for language_server_id in language_server_ids {
2113 if let Some(LanguageServerState::Running {
2114 adapter,
2115 simulate_disk_based_diagnostics_completion,
2116 ..
2117 }) = self.language_servers.get_mut(&language_server_id)
2118 {
2119 // After saving a buffer using a language server that doesn't provide
2120 // a disk-based progress token, kick off a timer that will reset every
2121 // time the buffer is saved. If the timer eventually fires, simulate
2122 // disk-based diagnostics being finished so that other pieces of UI
2123 // (e.g., project diagnostics view, diagnostic status bar) can update.
2124 // We don't emit an event right away because the language server might take
2125 // some time to publish diagnostics.
2126 if adapter.disk_based_diagnostics_progress_token.is_none() {
2127 const DISK_BASED_DIAGNOSTICS_DEBOUNCE: Duration =
2128 Duration::from_secs(1);
2129
2130 let task = cx.spawn_weak(|this, mut cx| async move {
2131 cx.background().timer(DISK_BASED_DIAGNOSTICS_DEBOUNCE).await;
2132 if let Some(this) = this.upgrade(&cx) {
2133 this.update(&mut cx, |this, cx| {
2134 this.disk_based_diagnostics_finished(
2135 language_server_id,
2136 cx,
2137 );
2138 this.buffer_ordered_messages_tx
2139 .unbounded_send(
2140 BufferOrderedMessage::LanguageServerUpdate {
2141 language_server_id,
2142 message:proto::update_language_server::Variant::DiskBasedDiagnosticsUpdated(Default::default())
2143 },
2144 )
2145 .ok();
2146 });
2147 }
2148 });
2149 *simulate_disk_based_diagnostics_completion = Some(task);
2150 }
2151 }
2152 }
2153 }
2154
2155 _ => {}
2156 }
2157
2158 None
2159 }
2160
2161 fn request_buffer_diff_recalculation(
2162 &mut self,
2163 buffer: &ModelHandle<Buffer>,
2164 cx: &mut ModelContext<Self>,
2165 ) {
2166 self.buffers_needing_diff.insert(buffer.downgrade());
2167 let first_insertion = self.buffers_needing_diff.len() == 1;
2168
2169 let settings = settings::get::<ProjectSettings>(cx);
2170 let delay = if let Some(delay) = settings.git.gutter_debounce {
2171 delay
2172 } else {
2173 if first_insertion {
2174 let this = cx.weak_handle();
2175 cx.defer(move |cx| {
2176 if let Some(this) = this.upgrade(cx) {
2177 this.update(cx, |this, cx| {
2178 this.recalculate_buffer_diffs(cx).detach();
2179 });
2180 }
2181 });
2182 }
2183 return;
2184 };
2185
2186 const MIN_DELAY: u64 = 50;
2187 let delay = delay.max(MIN_DELAY);
2188 let duration = Duration::from_millis(delay);
2189
2190 self.git_diff_debouncer
2191 .fire_new(duration, cx, move |this, cx| {
2192 this.recalculate_buffer_diffs(cx)
2193 });
2194 }
2195
2196 fn recalculate_buffer_diffs(&mut self, cx: &mut ModelContext<Self>) -> Task<()> {
2197 cx.spawn(|this, mut cx| async move {
2198 let buffers: Vec<_> = this.update(&mut cx, |this, _| {
2199 this.buffers_needing_diff.drain().collect()
2200 });
2201
2202 let tasks: Vec<_> = this.update(&mut cx, |_, cx| {
2203 buffers
2204 .iter()
2205 .filter_map(|buffer| {
2206 let buffer = buffer.upgrade(cx)?;
2207 buffer.update(cx, |buffer, cx| buffer.git_diff_recalc(cx))
2208 })
2209 .collect()
2210 });
2211
2212 futures::future::join_all(tasks).await;
2213
2214 this.update(&mut cx, |this, cx| {
2215 if !this.buffers_needing_diff.is_empty() {
2216 this.recalculate_buffer_diffs(cx).detach();
2217 } else {
2218 // TODO: Would a `ModelContext<Project>.notify()` suffice here?
2219 for buffer in buffers {
2220 if let Some(buffer) = buffer.upgrade(cx) {
2221 buffer.update(cx, |_, cx| cx.notify());
2222 }
2223 }
2224 }
2225 });
2226 })
2227 }
2228
2229 fn language_servers_for_worktree(
2230 &self,
2231 worktree_id: WorktreeId,
2232 ) -> impl Iterator<Item = (&Arc<CachedLspAdapter>, &Arc<Language>, &Arc<LanguageServer>)> {
2233 self.language_server_ids
2234 .iter()
2235 .filter_map(move |((language_server_worktree_id, _), id)| {
2236 if *language_server_worktree_id == worktree_id {
2237 if let Some(LanguageServerState::Running {
2238 adapter,
2239 language,
2240 server,
2241 ..
2242 }) = self.language_servers.get(id)
2243 {
2244 return Some((adapter, language, server));
2245 }
2246 }
2247 None
2248 })
2249 }
2250
2251 fn maintain_buffer_languages(
2252 languages: Arc<LanguageRegistry>,
2253 cx: &mut ModelContext<Project>,
2254 ) -> Task<()> {
2255 let mut subscription = languages.subscribe();
2256 let mut prev_reload_count = languages.reload_count();
2257 cx.spawn_weak(|project, mut cx| async move {
2258 while let Some(()) = subscription.next().await {
2259 if let Some(project) = project.upgrade(&cx) {
2260 // If the language registry has been reloaded, then remove and
2261 // re-assign the languages on all open buffers.
2262 let reload_count = languages.reload_count();
2263 if reload_count > prev_reload_count {
2264 prev_reload_count = reload_count;
2265 project.update(&mut cx, |this, cx| {
2266 let buffers = this
2267 .opened_buffers
2268 .values()
2269 .filter_map(|b| b.upgrade(cx))
2270 .collect::<Vec<_>>();
2271 for buffer in buffers {
2272 if let Some(f) = File::from_dyn(buffer.read(cx).file()).cloned() {
2273 this.unregister_buffer_from_language_servers(&buffer, &f, cx);
2274 buffer.update(cx, |buffer, cx| buffer.set_language(None, cx));
2275 }
2276 }
2277 });
2278 }
2279
2280 project.update(&mut cx, |project, cx| {
2281 let mut plain_text_buffers = Vec::new();
2282 let mut buffers_with_unknown_injections = Vec::new();
2283 for buffer in project.opened_buffers.values() {
2284 if let Some(handle) = buffer.upgrade(cx) {
2285 let buffer = &handle.read(cx);
2286 if buffer.language().is_none()
2287 || buffer.language() == Some(&*language::PLAIN_TEXT)
2288 {
2289 plain_text_buffers.push(handle);
2290 } else if buffer.contains_unknown_injections() {
2291 buffers_with_unknown_injections.push(handle);
2292 }
2293 }
2294 }
2295
2296 for buffer in plain_text_buffers {
2297 project.detect_language_for_buffer(&buffer, cx);
2298 project.register_buffer_with_language_servers(&buffer, cx);
2299 }
2300
2301 for buffer in buffers_with_unknown_injections {
2302 buffer.update(cx, |buffer, cx| buffer.reparse(cx));
2303 }
2304 });
2305 }
2306 }
2307 })
2308 }
2309
2310 fn maintain_workspace_config(
2311 languages: Arc<LanguageRegistry>,
2312 cx: &mut ModelContext<Project>,
2313 ) -> Task<()> {
2314 let (mut settings_changed_tx, mut settings_changed_rx) = watch::channel();
2315 let _ = postage::stream::Stream::try_recv(&mut settings_changed_rx);
2316
2317 let settings_observation = cx.observe_global::<SettingsStore, _>(move |_, _| {
2318 *settings_changed_tx.borrow_mut() = ();
2319 });
2320 cx.spawn_weak(|this, mut cx| async move {
2321 while let Some(_) = settings_changed_rx.next().await {
2322 let workspace_config = cx.update(|cx| languages.workspace_configuration(cx)).await;
2323 if let Some(this) = this.upgrade(&cx) {
2324 this.read_with(&cx, |this, _| {
2325 for server_state in this.language_servers.values() {
2326 if let LanguageServerState::Running { server, .. } = server_state {
2327 server
2328 .notify::<lsp::notification::DidChangeConfiguration>(
2329 lsp::DidChangeConfigurationParams {
2330 settings: workspace_config.clone(),
2331 },
2332 )
2333 .ok();
2334 }
2335 }
2336 })
2337 } else {
2338 break;
2339 }
2340 }
2341
2342 drop(settings_observation);
2343 })
2344 }
2345
2346 fn detect_language_for_buffer(
2347 &mut self,
2348 buffer_handle: &ModelHandle<Buffer>,
2349 cx: &mut ModelContext<Self>,
2350 ) -> Option<()> {
2351 // If the buffer has a language, set it and start the language server if we haven't already.
2352 let buffer = buffer_handle.read(cx);
2353 let full_path = buffer.file()?.full_path(cx);
2354 let content = buffer.as_rope();
2355 let new_language = self
2356 .languages
2357 .language_for_file(&full_path, Some(content))
2358 .now_or_never()?
2359 .ok()?;
2360 self.set_language_for_buffer(buffer_handle, new_language, cx);
2361 None
2362 }
2363
2364 pub fn set_language_for_buffer(
2365 &mut self,
2366 buffer: &ModelHandle<Buffer>,
2367 new_language: Arc<Language>,
2368 cx: &mut ModelContext<Self>,
2369 ) {
2370 buffer.update(cx, |buffer, cx| {
2371 if buffer.language().map_or(true, |old_language| {
2372 !Arc::ptr_eq(old_language, &new_language)
2373 }) {
2374 buffer.set_language(Some(new_language.clone()), cx);
2375 }
2376 });
2377
2378 if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
2379 let worktree = file.worktree.clone();
2380 if let Some(tree) = worktree.read(cx).as_local() {
2381 self.start_language_servers(&worktree, tree.abs_path().clone(), new_language, cx);
2382 }
2383 }
2384 }
2385
2386 fn start_language_servers(
2387 &mut self,
2388 worktree: &ModelHandle<Worktree>,
2389 worktree_path: Arc<Path>,
2390 language: Arc<Language>,
2391 cx: &mut ModelContext<Self>,
2392 ) {
2393 if !language_settings(
2394 Some(&language),
2395 worktree
2396 .update(cx, |tree, cx| tree.root_file(cx))
2397 .map(|f| f as _)
2398 .as_ref(),
2399 cx,
2400 )
2401 .enable_language_server
2402 {
2403 return;
2404 }
2405
2406 let worktree_id = worktree.read(cx).id();
2407 for adapter in language.lsp_adapters() {
2408 let key = (worktree_id, adapter.name.clone());
2409 if self.language_server_ids.contains_key(&key) {
2410 continue;
2411 }
2412
2413 let pending_server = match self.languages.start_language_server(
2414 language.clone(),
2415 adapter.clone(),
2416 worktree_path.clone(),
2417 ProjectLspAdapterDelegate::new(self, cx),
2418 cx,
2419 ) {
2420 Some(pending_server) => pending_server,
2421 None => continue,
2422 };
2423
2424 let lsp = settings::get::<ProjectSettings>(cx)
2425 .lsp
2426 .get(&adapter.name.0);
2427 let override_options = lsp.map(|s| s.initialization_options.clone()).flatten();
2428
2429 let mut initialization_options = adapter.initialization_options.clone();
2430 match (&mut initialization_options, override_options) {
2431 (Some(initialization_options), Some(override_options)) => {
2432 merge_json_value_into(override_options, initialization_options);
2433 }
2434 (None, override_options) => initialization_options = override_options,
2435 _ => {}
2436 }
2437
2438 let server_id = pending_server.server_id;
2439 let state = self.setup_pending_language_server(
2440 initialization_options,
2441 pending_server,
2442 adapter.clone(),
2443 language.clone(),
2444 key.clone(),
2445 cx,
2446 );
2447 self.language_servers.insert(server_id, state);
2448 self.language_server_ids.insert(key.clone(), server_id);
2449 }
2450 }
2451
2452 fn setup_pending_language_server(
2453 &mut self,
2454 initialization_options: Option<serde_json::Value>,
2455 pending_server: PendingLanguageServer,
2456 adapter: Arc<CachedLspAdapter>,
2457 language: Arc<Language>,
2458 key: (WorktreeId, LanguageServerName),
2459 cx: &mut ModelContext<Project>,
2460 ) -> LanguageServerState {
2461 let server_id = pending_server.server_id;
2462 let languages = self.languages.clone();
2463
2464 LanguageServerState::Starting(cx.spawn_weak(|this, mut cx| async move {
2465 let workspace_config = cx.update(|cx| languages.workspace_configuration(cx)).await;
2466 let language_server = pending_server.task.await.log_err()?;
2467
2468 language_server
2469 .on_notification::<lsp::notification::LogMessage, _>({
2470 move |params, mut cx| {
2471 if let Some(this) = this.upgrade(&cx) {
2472 this.update(&mut cx, |_, cx| {
2473 cx.emit(Event::LanguageServerLog(server_id, params.message))
2474 });
2475 }
2476 }
2477 })
2478 .detach();
2479
2480 language_server
2481 .on_notification::<lsp::notification::PublishDiagnostics, _>({
2482 let adapter = adapter.clone();
2483 move |mut params, cx| {
2484 let adapter = adapter.clone();
2485 cx.spawn(|mut cx| async move {
2486 adapter.process_diagnostics(&mut params).await;
2487 if let Some(this) = this.upgrade(&cx) {
2488 this.update(&mut cx, |this, cx| {
2489 this.update_diagnostics(
2490 server_id,
2491 params,
2492 &adapter.disk_based_diagnostic_sources,
2493 cx,
2494 )
2495 .log_err();
2496 });
2497 }
2498 })
2499 .detach();
2500 }
2501 })
2502 .detach();
2503
2504 language_server
2505 .on_request::<lsp::request::WorkspaceConfiguration, _, _>({
2506 let languages = languages.clone();
2507 move |params, mut cx| {
2508 let languages = languages.clone();
2509 async move {
2510 let workspace_config =
2511 cx.update(|cx| languages.workspace_configuration(cx)).await;
2512 Ok(params
2513 .items
2514 .into_iter()
2515 .map(|item| {
2516 if let Some(section) = &item.section {
2517 workspace_config
2518 .get(section)
2519 .cloned()
2520 .unwrap_or(serde_json::Value::Null)
2521 } else {
2522 workspace_config.clone()
2523 }
2524 })
2525 .collect())
2526 }
2527 }
2528 })
2529 .detach();
2530
2531 // Even though we don't have handling for these requests, respond to them to
2532 // avoid stalling any language server like `gopls` which waits for a response
2533 // to these requests when initializing.
2534 language_server
2535 .on_request::<lsp::request::WorkDoneProgressCreate, _, _>(
2536 move |params, mut cx| async move {
2537 if let Some(this) = this.upgrade(&cx) {
2538 this.update(&mut cx, |this, _| {
2539 if let Some(status) =
2540 this.language_server_statuses.get_mut(&server_id)
2541 {
2542 if let lsp::NumberOrString::String(token) = params.token {
2543 status.progress_tokens.insert(token);
2544 }
2545 }
2546 });
2547 }
2548 Ok(())
2549 },
2550 )
2551 .detach();
2552 language_server
2553 .on_request::<lsp::request::RegisterCapability, _, _>(
2554 move |params, mut cx| async move {
2555 let this = this
2556 .upgrade(&cx)
2557 .ok_or_else(|| anyhow!("project dropped"))?;
2558 for reg in params.registrations {
2559 if reg.method == "workspace/didChangeWatchedFiles" {
2560 if let Some(options) = reg.register_options {
2561 let options = serde_json::from_value(options)?;
2562 this.update(&mut cx, |this, cx| {
2563 this.on_lsp_did_change_watched_files(
2564 server_id, options, cx,
2565 );
2566 });
2567 }
2568 }
2569 }
2570 Ok(())
2571 },
2572 )
2573 .detach();
2574
2575 language_server
2576 .on_request::<lsp::request::ApplyWorkspaceEdit, _, _>({
2577 let adapter = adapter.clone();
2578 move |params, cx| {
2579 Self::on_lsp_workspace_edit(this, params, server_id, adapter.clone(), cx)
2580 }
2581 })
2582 .detach();
2583
2584 let disk_based_diagnostics_progress_token =
2585 adapter.disk_based_diagnostics_progress_token.clone();
2586
2587 language_server
2588 .on_notification::<lsp::notification::Progress, _>({
2589 move |params, mut cx| {
2590 if let Some(this) = this.upgrade(&cx) {
2591 this.update(&mut cx, |this, cx| {
2592 this.on_lsp_progress(
2593 params,
2594 server_id,
2595 disk_based_diagnostics_progress_token.clone(),
2596 cx,
2597 );
2598 });
2599 }
2600 }
2601 })
2602 .detach();
2603
2604 let language_server = language_server
2605 .initialize(initialization_options)
2606 .await
2607 .log_err()?;
2608 language_server
2609 .notify::<lsp::notification::DidChangeConfiguration>(
2610 lsp::DidChangeConfigurationParams {
2611 settings: workspace_config,
2612 },
2613 )
2614 .ok();
2615
2616 let this = this.upgrade(&cx)?;
2617 this.update(&mut cx, |this, cx| {
2618 // If the language server for this key doesn't match the server id, don't store the
2619 // server. Which will cause it to be dropped, killing the process
2620 if this
2621 .language_server_ids
2622 .get(&key)
2623 .map(|id| id != &server_id)
2624 .unwrap_or(false)
2625 {
2626 return None;
2627 }
2628
2629 // Update language_servers collection with Running variant of LanguageServerState
2630 // indicating that the server is up and running and ready
2631 this.language_servers.insert(
2632 server_id,
2633 LanguageServerState::Running {
2634 adapter: adapter.clone(),
2635 language: language.clone(),
2636 watched_paths: Default::default(),
2637 server: language_server.clone(),
2638 simulate_disk_based_diagnostics_completion: None,
2639 },
2640 );
2641 this.language_server_statuses.insert(
2642 server_id,
2643 LanguageServerStatus {
2644 name: language_server.name().to_string(),
2645 pending_work: Default::default(),
2646 has_pending_diagnostic_updates: false,
2647 progress_tokens: Default::default(),
2648 },
2649 );
2650
2651 cx.emit(Event::LanguageServerAdded(server_id));
2652
2653 if let Some(project_id) = this.remote_id() {
2654 this.client
2655 .send(proto::StartLanguageServer {
2656 project_id,
2657 server: Some(proto::LanguageServer {
2658 id: server_id.0 as u64,
2659 name: language_server.name().to_string(),
2660 }),
2661 })
2662 .log_err();
2663 }
2664
2665 // Tell the language server about every open buffer in the worktree that matches the language.
2666 for buffer in this.opened_buffers.values() {
2667 if let Some(buffer_handle) = buffer.upgrade(cx) {
2668 let buffer = buffer_handle.read(cx);
2669 let file = match File::from_dyn(buffer.file()) {
2670 Some(file) => file,
2671 None => continue,
2672 };
2673 let language = match buffer.language() {
2674 Some(language) => language,
2675 None => continue,
2676 };
2677
2678 if file.worktree.read(cx).id() != key.0
2679 || !language.lsp_adapters().iter().any(|a| a.name == key.1)
2680 {
2681 continue;
2682 }
2683
2684 let file = file.as_local()?;
2685 let versions = this
2686 .buffer_snapshots
2687 .entry(buffer.remote_id())
2688 .or_default()
2689 .entry(server_id)
2690 .or_insert_with(|| {
2691 vec![LspBufferSnapshot {
2692 version: 0,
2693 snapshot: buffer.text_snapshot(),
2694 }]
2695 });
2696
2697 let snapshot = versions.last().unwrap();
2698 let version = snapshot.version;
2699 let initial_snapshot = &snapshot.snapshot;
2700 let uri = lsp::Url::from_file_path(file.abs_path(cx)).unwrap();
2701 language_server
2702 .notify::<lsp::notification::DidOpenTextDocument>(
2703 lsp::DidOpenTextDocumentParams {
2704 text_document: lsp::TextDocumentItem::new(
2705 uri,
2706 adapter
2707 .language_ids
2708 .get(language.name().as_ref())
2709 .cloned()
2710 .unwrap_or_default(),
2711 version,
2712 initial_snapshot.text(),
2713 ),
2714 },
2715 )
2716 .log_err()?;
2717 buffer_handle.update(cx, |buffer, cx| {
2718 buffer.set_completion_triggers(
2719 language_server
2720 .capabilities()
2721 .completion_provider
2722 .as_ref()
2723 .and_then(|provider| provider.trigger_characters.clone())
2724 .unwrap_or_default(),
2725 cx,
2726 )
2727 });
2728 }
2729 }
2730
2731 cx.notify();
2732 Some(language_server)
2733 })
2734 }))
2735 }
2736
2737 // Returns a list of all of the worktrees which no longer have a language server and the root path
2738 // for the stopped server
2739 fn stop_language_server(
2740 &mut self,
2741 worktree_id: WorktreeId,
2742 adapter_name: LanguageServerName,
2743 cx: &mut ModelContext<Self>,
2744 ) -> Task<(Option<PathBuf>, Vec<WorktreeId>)> {
2745 let key = (worktree_id, adapter_name);
2746 if let Some(server_id) = self.language_server_ids.remove(&key) {
2747 // Remove other entries for this language server as well
2748 let mut orphaned_worktrees = vec![worktree_id];
2749 let other_keys = self.language_server_ids.keys().cloned().collect::<Vec<_>>();
2750 for other_key in other_keys {
2751 if self.language_server_ids.get(&other_key) == Some(&server_id) {
2752 self.language_server_ids.remove(&other_key);
2753 orphaned_worktrees.push(other_key.0);
2754 }
2755 }
2756
2757 for buffer in self.opened_buffers.values() {
2758 if let Some(buffer) = buffer.upgrade(cx) {
2759 buffer.update(cx, |buffer, cx| {
2760 buffer.update_diagnostics(server_id, Default::default(), cx);
2761 });
2762 }
2763 }
2764 for worktree in &self.worktrees {
2765 if let Some(worktree) = worktree.upgrade(cx) {
2766 worktree.update(cx, |worktree, cx| {
2767 if let Some(worktree) = worktree.as_local_mut() {
2768 worktree.clear_diagnostics_for_language_server(server_id, cx);
2769 }
2770 });
2771 }
2772 }
2773
2774 self.language_server_statuses.remove(&server_id);
2775 cx.notify();
2776
2777 let server_state = self.language_servers.remove(&server_id);
2778 cx.emit(Event::LanguageServerRemoved(server_id));
2779 cx.spawn_weak(|this, mut cx| async move {
2780 let mut root_path = None;
2781
2782 let server = match server_state {
2783 Some(LanguageServerState::Starting(started_language_server)) => {
2784 started_language_server.await
2785 }
2786 Some(LanguageServerState::Running { server, .. }) => Some(server),
2787 None => None,
2788 };
2789
2790 if let Some(server) = server {
2791 root_path = Some(server.root_path().clone());
2792 if let Some(shutdown) = server.shutdown() {
2793 shutdown.await;
2794 }
2795 }
2796
2797 if let Some(this) = this.upgrade(&cx) {
2798 this.update(&mut cx, |this, cx| {
2799 this.language_server_statuses.remove(&server_id);
2800 cx.notify();
2801 });
2802 }
2803
2804 (root_path, orphaned_worktrees)
2805 })
2806 } else {
2807 Task::ready((None, Vec::new()))
2808 }
2809 }
2810
2811 pub fn restart_language_servers_for_buffers(
2812 &mut self,
2813 buffers: impl IntoIterator<Item = ModelHandle<Buffer>>,
2814 cx: &mut ModelContext<Self>,
2815 ) -> Option<()> {
2816 let language_server_lookup_info: HashSet<(ModelHandle<Worktree>, Arc<Language>)> = buffers
2817 .into_iter()
2818 .filter_map(|buffer| {
2819 let buffer = buffer.read(cx);
2820 let file = File::from_dyn(buffer.file())?;
2821 let full_path = file.full_path(cx);
2822 let language = self
2823 .languages
2824 .language_for_file(&full_path, Some(buffer.as_rope()))
2825 .now_or_never()?
2826 .ok()?;
2827 Some((file.worktree.clone(), language))
2828 })
2829 .collect();
2830 for (worktree, language) in language_server_lookup_info {
2831 self.restart_language_servers(worktree, language, cx);
2832 }
2833
2834 None
2835 }
2836
2837 // TODO This will break in the case where the adapter's root paths and worktrees are not equal
2838 fn restart_language_servers(
2839 &mut self,
2840 worktree: ModelHandle<Worktree>,
2841 language: Arc<Language>,
2842 cx: &mut ModelContext<Self>,
2843 ) {
2844 let worktree_id = worktree.read(cx).id();
2845 let fallback_path = worktree.read(cx).abs_path();
2846
2847 let mut stops = Vec::new();
2848 for adapter in language.lsp_adapters() {
2849 stops.push(self.stop_language_server(worktree_id, adapter.name.clone(), cx));
2850 }
2851
2852 if stops.is_empty() {
2853 return;
2854 }
2855 let mut stops = stops.into_iter();
2856
2857 cx.spawn_weak(|this, mut cx| async move {
2858 let (original_root_path, mut orphaned_worktrees) = stops.next().unwrap().await;
2859 for stop in stops {
2860 let (_, worktrees) = stop.await;
2861 orphaned_worktrees.extend_from_slice(&worktrees);
2862 }
2863
2864 let this = match this.upgrade(&cx) {
2865 Some(this) => this,
2866 None => return,
2867 };
2868
2869 this.update(&mut cx, |this, cx| {
2870 // Attempt to restart using original server path. Fallback to passed in
2871 // path if we could not retrieve the root path
2872 let root_path = original_root_path
2873 .map(|path_buf| Arc::from(path_buf.as_path()))
2874 .unwrap_or(fallback_path);
2875
2876 this.start_language_servers(&worktree, root_path, language.clone(), cx);
2877
2878 // Lookup new server ids and set them for each of the orphaned worktrees
2879 for adapter in language.lsp_adapters() {
2880 if let Some(new_server_id) = this
2881 .language_server_ids
2882 .get(&(worktree_id, adapter.name.clone()))
2883 .cloned()
2884 {
2885 for &orphaned_worktree in &orphaned_worktrees {
2886 this.language_server_ids
2887 .insert((orphaned_worktree, adapter.name.clone()), new_server_id);
2888 }
2889 }
2890 }
2891 });
2892 })
2893 .detach();
2894 }
2895
2896 fn on_lsp_progress(
2897 &mut self,
2898 progress: lsp::ProgressParams,
2899 language_server_id: LanguageServerId,
2900 disk_based_diagnostics_progress_token: Option<String>,
2901 cx: &mut ModelContext<Self>,
2902 ) {
2903 let token = match progress.token {
2904 lsp::NumberOrString::String(token) => token,
2905 lsp::NumberOrString::Number(token) => {
2906 log::info!("skipping numeric progress token {}", token);
2907 return;
2908 }
2909 };
2910 let lsp::ProgressParamsValue::WorkDone(progress) = progress.value;
2911 let language_server_status =
2912 if let Some(status) = self.language_server_statuses.get_mut(&language_server_id) {
2913 status
2914 } else {
2915 return;
2916 };
2917
2918 if !language_server_status.progress_tokens.contains(&token) {
2919 return;
2920 }
2921
2922 let is_disk_based_diagnostics_progress = disk_based_diagnostics_progress_token
2923 .as_ref()
2924 .map_or(false, |disk_based_token| {
2925 token.starts_with(disk_based_token)
2926 });
2927
2928 match progress {
2929 lsp::WorkDoneProgress::Begin(report) => {
2930 if is_disk_based_diagnostics_progress {
2931 language_server_status.has_pending_diagnostic_updates = true;
2932 self.disk_based_diagnostics_started(language_server_id, cx);
2933 self.buffer_ordered_messages_tx
2934 .unbounded_send(BufferOrderedMessage::LanguageServerUpdate {
2935 language_server_id,
2936 message: proto::update_language_server::Variant::DiskBasedDiagnosticsUpdating(Default::default())
2937 })
2938 .ok();
2939 } else {
2940 self.on_lsp_work_start(
2941 language_server_id,
2942 token.clone(),
2943 LanguageServerProgress {
2944 message: report.message.clone(),
2945 percentage: report.percentage.map(|p| p as usize),
2946 last_update_at: Instant::now(),
2947 },
2948 cx,
2949 );
2950 self.buffer_ordered_messages_tx
2951 .unbounded_send(BufferOrderedMessage::LanguageServerUpdate {
2952 language_server_id,
2953 message: proto::update_language_server::Variant::WorkStart(
2954 proto::LspWorkStart {
2955 token,
2956 message: report.message,
2957 percentage: report.percentage.map(|p| p as u32),
2958 },
2959 ),
2960 })
2961 .ok();
2962 }
2963 }
2964 lsp::WorkDoneProgress::Report(report) => {
2965 if !is_disk_based_diagnostics_progress {
2966 self.on_lsp_work_progress(
2967 language_server_id,
2968 token.clone(),
2969 LanguageServerProgress {
2970 message: report.message.clone(),
2971 percentage: report.percentage.map(|p| p as usize),
2972 last_update_at: Instant::now(),
2973 },
2974 cx,
2975 );
2976 self.buffer_ordered_messages_tx
2977 .unbounded_send(BufferOrderedMessage::LanguageServerUpdate {
2978 language_server_id,
2979 message: proto::update_language_server::Variant::WorkProgress(
2980 proto::LspWorkProgress {
2981 token,
2982 message: report.message,
2983 percentage: report.percentage.map(|p| p as u32),
2984 },
2985 ),
2986 })
2987 .ok();
2988 }
2989 }
2990 lsp::WorkDoneProgress::End(_) => {
2991 language_server_status.progress_tokens.remove(&token);
2992
2993 if is_disk_based_diagnostics_progress {
2994 language_server_status.has_pending_diagnostic_updates = false;
2995 self.disk_based_diagnostics_finished(language_server_id, cx);
2996 self.buffer_ordered_messages_tx
2997 .unbounded_send(BufferOrderedMessage::LanguageServerUpdate {
2998 language_server_id,
2999 message:
3000 proto::update_language_server::Variant::DiskBasedDiagnosticsUpdated(
3001 Default::default(),
3002 ),
3003 })
3004 .ok();
3005 } else {
3006 self.on_lsp_work_end(language_server_id, token.clone(), cx);
3007 self.buffer_ordered_messages_tx
3008 .unbounded_send(BufferOrderedMessage::LanguageServerUpdate {
3009 language_server_id,
3010 message: proto::update_language_server::Variant::WorkEnd(
3011 proto::LspWorkEnd { token },
3012 ),
3013 })
3014 .ok();
3015 }
3016 }
3017 }
3018 }
3019
3020 fn on_lsp_work_start(
3021 &mut self,
3022 language_server_id: LanguageServerId,
3023 token: String,
3024 progress: LanguageServerProgress,
3025 cx: &mut ModelContext<Self>,
3026 ) {
3027 if let Some(status) = self.language_server_statuses.get_mut(&language_server_id) {
3028 status.pending_work.insert(token, progress);
3029 cx.notify();
3030 }
3031 }
3032
3033 fn on_lsp_work_progress(
3034 &mut self,
3035 language_server_id: LanguageServerId,
3036 token: String,
3037 progress: LanguageServerProgress,
3038 cx: &mut ModelContext<Self>,
3039 ) {
3040 if let Some(status) = self.language_server_statuses.get_mut(&language_server_id) {
3041 let entry = status
3042 .pending_work
3043 .entry(token)
3044 .or_insert(LanguageServerProgress {
3045 message: Default::default(),
3046 percentage: Default::default(),
3047 last_update_at: progress.last_update_at,
3048 });
3049 if progress.message.is_some() {
3050 entry.message = progress.message;
3051 }
3052 if progress.percentage.is_some() {
3053 entry.percentage = progress.percentage;
3054 }
3055 entry.last_update_at = progress.last_update_at;
3056 cx.notify();
3057 }
3058 }
3059
3060 fn on_lsp_work_end(
3061 &mut self,
3062 language_server_id: LanguageServerId,
3063 token: String,
3064 cx: &mut ModelContext<Self>,
3065 ) {
3066 if let Some(status) = self.language_server_statuses.get_mut(&language_server_id) {
3067 status.pending_work.remove(&token);
3068 cx.notify();
3069 }
3070 }
3071
3072 fn on_lsp_did_change_watched_files(
3073 &mut self,
3074 language_server_id: LanguageServerId,
3075 params: DidChangeWatchedFilesRegistrationOptions,
3076 cx: &mut ModelContext<Self>,
3077 ) {
3078 if let Some(LanguageServerState::Running { watched_paths, .. }) =
3079 self.language_servers.get_mut(&language_server_id)
3080 {
3081 let mut builders = HashMap::default();
3082 for watcher in params.watchers {
3083 for worktree in &self.worktrees {
3084 if let Some(worktree) = worktree.upgrade(cx) {
3085 let worktree = worktree.read(cx);
3086 if let Some(abs_path) = worktree.abs_path().to_str() {
3087 if let Some(suffix) = match &watcher.glob_pattern {
3088 lsp::GlobPattern::String(s) => s,
3089 lsp::GlobPattern::Relative(rp) => &rp.pattern,
3090 }
3091 .strip_prefix(abs_path)
3092 .and_then(|s| s.strip_prefix(std::path::MAIN_SEPARATOR))
3093 {
3094 if let Some(glob) = Glob::new(suffix).log_err() {
3095 builders
3096 .entry(worktree.id())
3097 .or_insert_with(|| GlobSetBuilder::new())
3098 .add(glob);
3099 }
3100 break;
3101 }
3102 }
3103 }
3104 }
3105 }
3106
3107 watched_paths.clear();
3108 for (worktree_id, builder) in builders {
3109 if let Ok(globset) = builder.build() {
3110 watched_paths.insert(worktree_id, globset);
3111 }
3112 }
3113
3114 cx.notify();
3115 }
3116 }
3117
3118 async fn on_lsp_workspace_edit(
3119 this: WeakModelHandle<Self>,
3120 params: lsp::ApplyWorkspaceEditParams,
3121 server_id: LanguageServerId,
3122 adapter: Arc<CachedLspAdapter>,
3123 mut cx: AsyncAppContext,
3124 ) -> Result<lsp::ApplyWorkspaceEditResponse> {
3125 let this = this
3126 .upgrade(&cx)
3127 .ok_or_else(|| anyhow!("project project closed"))?;
3128 let language_server = this
3129 .read_with(&cx, |this, _| this.language_server_for_id(server_id))
3130 .ok_or_else(|| anyhow!("language server not found"))?;
3131 let transaction = Self::deserialize_workspace_edit(
3132 this.clone(),
3133 params.edit,
3134 true,
3135 adapter.clone(),
3136 language_server.clone(),
3137 &mut cx,
3138 )
3139 .await
3140 .log_err();
3141 this.update(&mut cx, |this, _| {
3142 if let Some(transaction) = transaction {
3143 this.last_workspace_edits_by_language_server
3144 .insert(server_id, transaction);
3145 }
3146 });
3147 Ok(lsp::ApplyWorkspaceEditResponse {
3148 applied: true,
3149 failed_change: None,
3150 failure_reason: None,
3151 })
3152 }
3153
3154 pub fn language_server_statuses(
3155 &self,
3156 ) -> impl DoubleEndedIterator<Item = &LanguageServerStatus> {
3157 self.language_server_statuses.values()
3158 }
3159
3160 pub fn update_diagnostics(
3161 &mut self,
3162 language_server_id: LanguageServerId,
3163 mut params: lsp::PublishDiagnosticsParams,
3164 disk_based_sources: &[String],
3165 cx: &mut ModelContext<Self>,
3166 ) -> Result<()> {
3167 let abs_path = params
3168 .uri
3169 .to_file_path()
3170 .map_err(|_| anyhow!("URI is not a file"))?;
3171 let mut diagnostics = Vec::default();
3172 let mut primary_diagnostic_group_ids = HashMap::default();
3173 let mut sources_by_group_id = HashMap::default();
3174 let mut supporting_diagnostics = HashMap::default();
3175
3176 // Ensure that primary diagnostics are always the most severe
3177 params.diagnostics.sort_by_key(|item| item.severity);
3178
3179 for diagnostic in ¶ms.diagnostics {
3180 let source = diagnostic.source.as_ref();
3181 let code = diagnostic.code.as_ref().map(|code| match code {
3182 lsp::NumberOrString::Number(code) => code.to_string(),
3183 lsp::NumberOrString::String(code) => code.clone(),
3184 });
3185 let range = range_from_lsp(diagnostic.range);
3186 let is_supporting = diagnostic
3187 .related_information
3188 .as_ref()
3189 .map_or(false, |infos| {
3190 infos.iter().any(|info| {
3191 primary_diagnostic_group_ids.contains_key(&(
3192 source,
3193 code.clone(),
3194 range_from_lsp(info.location.range),
3195 ))
3196 })
3197 });
3198
3199 let is_unnecessary = diagnostic.tags.as_ref().map_or(false, |tags| {
3200 tags.iter().any(|tag| *tag == DiagnosticTag::UNNECESSARY)
3201 });
3202
3203 if is_supporting {
3204 supporting_diagnostics.insert(
3205 (source, code.clone(), range),
3206 (diagnostic.severity, is_unnecessary),
3207 );
3208 } else {
3209 let group_id = post_inc(&mut self.next_diagnostic_group_id);
3210 let is_disk_based =
3211 source.map_or(false, |source| disk_based_sources.contains(source));
3212
3213 sources_by_group_id.insert(group_id, source);
3214 primary_diagnostic_group_ids
3215 .insert((source, code.clone(), range.clone()), group_id);
3216
3217 diagnostics.push(DiagnosticEntry {
3218 range,
3219 diagnostic: Diagnostic {
3220 source: diagnostic.source.clone(),
3221 code: code.clone(),
3222 severity: diagnostic.severity.unwrap_or(DiagnosticSeverity::ERROR),
3223 message: diagnostic.message.clone(),
3224 group_id,
3225 is_primary: true,
3226 is_valid: true,
3227 is_disk_based,
3228 is_unnecessary,
3229 },
3230 });
3231 if let Some(infos) = &diagnostic.related_information {
3232 for info in infos {
3233 if info.location.uri == params.uri && !info.message.is_empty() {
3234 let range = range_from_lsp(info.location.range);
3235 diagnostics.push(DiagnosticEntry {
3236 range,
3237 diagnostic: Diagnostic {
3238 source: diagnostic.source.clone(),
3239 code: code.clone(),
3240 severity: DiagnosticSeverity::INFORMATION,
3241 message: info.message.clone(),
3242 group_id,
3243 is_primary: false,
3244 is_valid: true,
3245 is_disk_based,
3246 is_unnecessary: false,
3247 },
3248 });
3249 }
3250 }
3251 }
3252 }
3253 }
3254
3255 for entry in &mut diagnostics {
3256 let diagnostic = &mut entry.diagnostic;
3257 if !diagnostic.is_primary {
3258 let source = *sources_by_group_id.get(&diagnostic.group_id).unwrap();
3259 if let Some(&(severity, is_unnecessary)) = supporting_diagnostics.get(&(
3260 source,
3261 diagnostic.code.clone(),
3262 entry.range.clone(),
3263 )) {
3264 if let Some(severity) = severity {
3265 diagnostic.severity = severity;
3266 }
3267 diagnostic.is_unnecessary = is_unnecessary;
3268 }
3269 }
3270 }
3271
3272 self.update_diagnostic_entries(
3273 language_server_id,
3274 abs_path,
3275 params.version,
3276 diagnostics,
3277 cx,
3278 )?;
3279 Ok(())
3280 }
3281
3282 pub fn update_diagnostic_entries(
3283 &mut self,
3284 server_id: LanguageServerId,
3285 abs_path: PathBuf,
3286 version: Option<i32>,
3287 diagnostics: Vec<DiagnosticEntry<Unclipped<PointUtf16>>>,
3288 cx: &mut ModelContext<Project>,
3289 ) -> Result<(), anyhow::Error> {
3290 let (worktree, relative_path) = self
3291 .find_local_worktree(&abs_path, cx)
3292 .ok_or_else(|| anyhow!("no worktree found for diagnostics path {abs_path:?}"))?;
3293
3294 let project_path = ProjectPath {
3295 worktree_id: worktree.read(cx).id(),
3296 path: relative_path.into(),
3297 };
3298
3299 if let Some(buffer) = self.get_open_buffer(&project_path, cx) {
3300 self.update_buffer_diagnostics(&buffer, server_id, version, diagnostics.clone(), cx)?;
3301 }
3302
3303 let updated = worktree.update(cx, |worktree, cx| {
3304 worktree
3305 .as_local_mut()
3306 .ok_or_else(|| anyhow!("not a local worktree"))?
3307 .update_diagnostics(server_id, project_path.path.clone(), diagnostics, cx)
3308 })?;
3309 if updated {
3310 cx.emit(Event::DiagnosticsUpdated {
3311 language_server_id: server_id,
3312 path: project_path,
3313 });
3314 }
3315 Ok(())
3316 }
3317
3318 fn update_buffer_diagnostics(
3319 &mut self,
3320 buffer: &ModelHandle<Buffer>,
3321 server_id: LanguageServerId,
3322 version: Option<i32>,
3323 mut diagnostics: Vec<DiagnosticEntry<Unclipped<PointUtf16>>>,
3324 cx: &mut ModelContext<Self>,
3325 ) -> Result<()> {
3326 fn compare_diagnostics(a: &Diagnostic, b: &Diagnostic) -> Ordering {
3327 Ordering::Equal
3328 .then_with(|| b.is_primary.cmp(&a.is_primary))
3329 .then_with(|| a.is_disk_based.cmp(&b.is_disk_based))
3330 .then_with(|| a.severity.cmp(&b.severity))
3331 .then_with(|| a.message.cmp(&b.message))
3332 }
3333
3334 let snapshot = self.buffer_snapshot_for_lsp_version(buffer, server_id, version, cx)?;
3335
3336 diagnostics.sort_unstable_by(|a, b| {
3337 Ordering::Equal
3338 .then_with(|| a.range.start.cmp(&b.range.start))
3339 .then_with(|| b.range.end.cmp(&a.range.end))
3340 .then_with(|| compare_diagnostics(&a.diagnostic, &b.diagnostic))
3341 });
3342
3343 let mut sanitized_diagnostics = Vec::new();
3344 let edits_since_save = Patch::new(
3345 snapshot
3346 .edits_since::<Unclipped<PointUtf16>>(buffer.read(cx).saved_version())
3347 .collect(),
3348 );
3349 for entry in diagnostics {
3350 let start;
3351 let end;
3352 if entry.diagnostic.is_disk_based {
3353 // Some diagnostics are based on files on disk instead of buffers'
3354 // current contents. Adjust these diagnostics' ranges to reflect
3355 // any unsaved edits.
3356 start = edits_since_save.old_to_new(entry.range.start);
3357 end = edits_since_save.old_to_new(entry.range.end);
3358 } else {
3359 start = entry.range.start;
3360 end = entry.range.end;
3361 }
3362
3363 let mut range = snapshot.clip_point_utf16(start, Bias::Left)
3364 ..snapshot.clip_point_utf16(end, Bias::Right);
3365
3366 // Expand empty ranges by one codepoint
3367 if range.start == range.end {
3368 // This will be go to the next boundary when being clipped
3369 range.end.column += 1;
3370 range.end = snapshot.clip_point_utf16(Unclipped(range.end), Bias::Right);
3371 if range.start == range.end && range.end.column > 0 {
3372 range.start.column -= 1;
3373 range.end = snapshot.clip_point_utf16(Unclipped(range.end), Bias::Left);
3374 }
3375 }
3376
3377 sanitized_diagnostics.push(DiagnosticEntry {
3378 range,
3379 diagnostic: entry.diagnostic,
3380 });
3381 }
3382 drop(edits_since_save);
3383
3384 let set = DiagnosticSet::new(sanitized_diagnostics, &snapshot);
3385 buffer.update(cx, |buffer, cx| {
3386 buffer.update_diagnostics(server_id, set, cx)
3387 });
3388 Ok(())
3389 }
3390
3391 pub fn reload_buffers(
3392 &self,
3393 buffers: HashSet<ModelHandle<Buffer>>,
3394 push_to_history: bool,
3395 cx: &mut ModelContext<Self>,
3396 ) -> Task<Result<ProjectTransaction>> {
3397 let mut local_buffers = Vec::new();
3398 let mut remote_buffers = None;
3399 for buffer_handle in buffers {
3400 let buffer = buffer_handle.read(cx);
3401 if buffer.is_dirty() {
3402 if let Some(file) = File::from_dyn(buffer.file()) {
3403 if file.is_local() {
3404 local_buffers.push(buffer_handle);
3405 } else {
3406 remote_buffers.get_or_insert(Vec::new()).push(buffer_handle);
3407 }
3408 }
3409 }
3410 }
3411
3412 let remote_buffers = self.remote_id().zip(remote_buffers);
3413 let client = self.client.clone();
3414
3415 cx.spawn(|this, mut cx| async move {
3416 let mut project_transaction = ProjectTransaction::default();
3417
3418 if let Some((project_id, remote_buffers)) = remote_buffers {
3419 let response = client
3420 .request(proto::ReloadBuffers {
3421 project_id,
3422 buffer_ids: remote_buffers
3423 .iter()
3424 .map(|buffer| buffer.read_with(&cx, |buffer, _| buffer.remote_id()))
3425 .collect(),
3426 })
3427 .await?
3428 .transaction
3429 .ok_or_else(|| anyhow!("missing transaction"))?;
3430 project_transaction = this
3431 .update(&mut cx, |this, cx| {
3432 this.deserialize_project_transaction(response, push_to_history, cx)
3433 })
3434 .await?;
3435 }
3436
3437 for buffer in local_buffers {
3438 let transaction = buffer
3439 .update(&mut cx, |buffer, cx| buffer.reload(cx))
3440 .await?;
3441 buffer.update(&mut cx, |buffer, cx| {
3442 if let Some(transaction) = transaction {
3443 if !push_to_history {
3444 buffer.forget_transaction(transaction.id);
3445 }
3446 project_transaction.0.insert(cx.handle(), transaction);
3447 }
3448 });
3449 }
3450
3451 Ok(project_transaction)
3452 })
3453 }
3454
3455 pub fn format(
3456 &self,
3457 buffers: HashSet<ModelHandle<Buffer>>,
3458 push_to_history: bool,
3459 trigger: FormatTrigger,
3460 cx: &mut ModelContext<Project>,
3461 ) -> Task<Result<ProjectTransaction>> {
3462 if self.is_local() {
3463 let mut buffers_with_paths_and_servers = buffers
3464 .into_iter()
3465 .filter_map(|buffer_handle| {
3466 let buffer = buffer_handle.read(cx);
3467 let file = File::from_dyn(buffer.file())?;
3468 let buffer_abs_path = file.as_local().map(|f| f.abs_path(cx));
3469 let server = self
3470 .primary_language_servers_for_buffer(buffer, cx)
3471 .map(|s| s.1.clone());
3472 Some((buffer_handle, buffer_abs_path, server))
3473 })
3474 .collect::<Vec<_>>();
3475
3476 cx.spawn(|this, mut cx| async move {
3477 // Do not allow multiple concurrent formatting requests for the
3478 // same buffer.
3479 this.update(&mut cx, |this, cx| {
3480 buffers_with_paths_and_servers.retain(|(buffer, _, _)| {
3481 this.buffers_being_formatted
3482 .insert(buffer.read(cx).remote_id())
3483 });
3484 });
3485
3486 let _cleanup = defer({
3487 let this = this.clone();
3488 let mut cx = cx.clone();
3489 let buffers = &buffers_with_paths_and_servers;
3490 move || {
3491 this.update(&mut cx, |this, cx| {
3492 for (buffer, _, _) in buffers {
3493 this.buffers_being_formatted
3494 .remove(&buffer.read(cx).remote_id());
3495 }
3496 });
3497 }
3498 });
3499
3500 let mut project_transaction = ProjectTransaction::default();
3501 for (buffer, buffer_abs_path, language_server) in &buffers_with_paths_and_servers {
3502 let settings = buffer.read_with(&cx, |buffer, cx| {
3503 language_settings(buffer.language(), buffer.file(), cx).clone()
3504 });
3505
3506 let remove_trailing_whitespace = settings.remove_trailing_whitespace_on_save;
3507 let ensure_final_newline = settings.ensure_final_newline_on_save;
3508 let format_on_save = settings.format_on_save.clone();
3509 let formatter = settings.formatter.clone();
3510 let tab_size = settings.tab_size;
3511
3512 // First, format buffer's whitespace according to the settings.
3513 let trailing_whitespace_diff = if remove_trailing_whitespace {
3514 Some(
3515 buffer
3516 .read_with(&cx, |b, cx| b.remove_trailing_whitespace(cx))
3517 .await,
3518 )
3519 } else {
3520 None
3521 };
3522 let whitespace_transaction_id = buffer.update(&mut cx, |buffer, cx| {
3523 buffer.finalize_last_transaction();
3524 buffer.start_transaction();
3525 if let Some(diff) = trailing_whitespace_diff {
3526 buffer.apply_diff(diff, cx);
3527 }
3528 if ensure_final_newline {
3529 buffer.ensure_final_newline(cx);
3530 }
3531 buffer.end_transaction(cx)
3532 });
3533
3534 // Currently, formatting operations are represented differently depending on
3535 // whether they come from a language server or an external command.
3536 enum FormatOperation {
3537 Lsp(Vec<(Range<Anchor>, String)>),
3538 External(Diff),
3539 }
3540
3541 // Apply language-specific formatting using either a language server
3542 // or external command.
3543 let mut format_operation = None;
3544 match (formatter, format_on_save) {
3545 (_, FormatOnSave::Off) if trigger == FormatTrigger::Save => {}
3546
3547 (Formatter::LanguageServer, FormatOnSave::On | FormatOnSave::Off)
3548 | (_, FormatOnSave::LanguageServer) => {
3549 if let Some((language_server, buffer_abs_path)) =
3550 language_server.as_ref().zip(buffer_abs_path.as_ref())
3551 {
3552 format_operation = Some(FormatOperation::Lsp(
3553 Self::format_via_lsp(
3554 &this,
3555 &buffer,
3556 buffer_abs_path,
3557 &language_server,
3558 tab_size,
3559 &mut cx,
3560 )
3561 .await
3562 .context("failed to format via language server")?,
3563 ));
3564 }
3565 }
3566
3567 (
3568 Formatter::External { command, arguments },
3569 FormatOnSave::On | FormatOnSave::Off,
3570 )
3571 | (_, FormatOnSave::External { command, arguments }) => {
3572 if let Some(buffer_abs_path) = buffer_abs_path {
3573 format_operation = Self::format_via_external_command(
3574 &buffer,
3575 &buffer_abs_path,
3576 &command,
3577 &arguments,
3578 &mut cx,
3579 )
3580 .await
3581 .context(format!(
3582 "failed to format via external command {:?}",
3583 command
3584 ))?
3585 .map(FormatOperation::External);
3586 }
3587 }
3588 };
3589
3590 buffer.update(&mut cx, |b, cx| {
3591 // If the buffer had its whitespace formatted and was edited while the language-specific
3592 // formatting was being computed, avoid applying the language-specific formatting, because
3593 // it can't be grouped with the whitespace formatting in the undo history.
3594 if let Some(transaction_id) = whitespace_transaction_id {
3595 if b.peek_undo_stack()
3596 .map_or(true, |e| e.transaction_id() != transaction_id)
3597 {
3598 format_operation.take();
3599 }
3600 }
3601
3602 // Apply any language-specific formatting, and group the two formatting operations
3603 // in the buffer's undo history.
3604 if let Some(operation) = format_operation {
3605 match operation {
3606 FormatOperation::Lsp(edits) => {
3607 b.edit(edits, None, cx);
3608 }
3609 FormatOperation::External(diff) => {
3610 b.apply_diff(diff, cx);
3611 }
3612 }
3613
3614 if let Some(transaction_id) = whitespace_transaction_id {
3615 b.group_until_transaction(transaction_id);
3616 }
3617 }
3618
3619 if let Some(transaction) = b.finalize_last_transaction().cloned() {
3620 if !push_to_history {
3621 b.forget_transaction(transaction.id);
3622 }
3623 project_transaction.0.insert(buffer.clone(), transaction);
3624 }
3625 });
3626 }
3627
3628 Ok(project_transaction)
3629 })
3630 } else {
3631 let remote_id = self.remote_id();
3632 let client = self.client.clone();
3633 cx.spawn(|this, mut cx| async move {
3634 let mut project_transaction = ProjectTransaction::default();
3635 if let Some(project_id) = remote_id {
3636 let response = client
3637 .request(proto::FormatBuffers {
3638 project_id,
3639 trigger: trigger as i32,
3640 buffer_ids: buffers
3641 .iter()
3642 .map(|buffer| buffer.read_with(&cx, |buffer, _| buffer.remote_id()))
3643 .collect(),
3644 })
3645 .await?
3646 .transaction
3647 .ok_or_else(|| anyhow!("missing transaction"))?;
3648 project_transaction = this
3649 .update(&mut cx, |this, cx| {
3650 this.deserialize_project_transaction(response, push_to_history, cx)
3651 })
3652 .await?;
3653 }
3654 Ok(project_transaction)
3655 })
3656 }
3657 }
3658
3659 async fn format_via_lsp(
3660 this: &ModelHandle<Self>,
3661 buffer: &ModelHandle<Buffer>,
3662 abs_path: &Path,
3663 language_server: &Arc<LanguageServer>,
3664 tab_size: NonZeroU32,
3665 cx: &mut AsyncAppContext,
3666 ) -> Result<Vec<(Range<Anchor>, String)>> {
3667 let text_document =
3668 lsp::TextDocumentIdentifier::new(lsp::Url::from_file_path(abs_path).unwrap());
3669 let capabilities = &language_server.capabilities();
3670 let lsp_edits = if capabilities
3671 .document_formatting_provider
3672 .as_ref()
3673 .map_or(false, |provider| *provider != lsp::OneOf::Left(false))
3674 {
3675 language_server
3676 .request::<lsp::request::Formatting>(lsp::DocumentFormattingParams {
3677 text_document,
3678 options: lsp_command::lsp_formatting_options(tab_size.get()),
3679 work_done_progress_params: Default::default(),
3680 })
3681 .await?
3682 } else if capabilities
3683 .document_range_formatting_provider
3684 .as_ref()
3685 .map_or(false, |provider| *provider != lsp::OneOf::Left(false))
3686 {
3687 let buffer_start = lsp::Position::new(0, 0);
3688 let buffer_end =
3689 buffer.read_with(cx, |buffer, _| point_to_lsp(buffer.max_point_utf16()));
3690 language_server
3691 .request::<lsp::request::RangeFormatting>(lsp::DocumentRangeFormattingParams {
3692 text_document,
3693 range: lsp::Range::new(buffer_start, buffer_end),
3694 options: lsp_command::lsp_formatting_options(tab_size.get()),
3695 work_done_progress_params: Default::default(),
3696 })
3697 .await?
3698 } else {
3699 None
3700 };
3701
3702 if let Some(lsp_edits) = lsp_edits {
3703 this.update(cx, |this, cx| {
3704 this.edits_from_lsp(buffer, lsp_edits, language_server.server_id(), None, cx)
3705 })
3706 .await
3707 } else {
3708 Ok(Default::default())
3709 }
3710 }
3711
3712 async fn format_via_external_command(
3713 buffer: &ModelHandle<Buffer>,
3714 buffer_abs_path: &Path,
3715 command: &str,
3716 arguments: &[String],
3717 cx: &mut AsyncAppContext,
3718 ) -> Result<Option<Diff>> {
3719 let working_dir_path = buffer.read_with(cx, |buffer, cx| {
3720 let file = File::from_dyn(buffer.file())?;
3721 let worktree = file.worktree.read(cx).as_local()?;
3722 let mut worktree_path = worktree.abs_path().to_path_buf();
3723 if worktree.root_entry()?.is_file() {
3724 worktree_path.pop();
3725 }
3726 Some(worktree_path)
3727 });
3728
3729 if let Some(working_dir_path) = working_dir_path {
3730 let mut child =
3731 smol::process::Command::new(command)
3732 .args(arguments.iter().map(|arg| {
3733 arg.replace("{buffer_path}", &buffer_abs_path.to_string_lossy())
3734 }))
3735 .current_dir(&working_dir_path)
3736 .stdin(smol::process::Stdio::piped())
3737 .stdout(smol::process::Stdio::piped())
3738 .stderr(smol::process::Stdio::piped())
3739 .spawn()?;
3740 let stdin = child
3741 .stdin
3742 .as_mut()
3743 .ok_or_else(|| anyhow!("failed to acquire stdin"))?;
3744 let text = buffer.read_with(cx, |buffer, _| buffer.as_rope().clone());
3745 for chunk in text.chunks() {
3746 stdin.write_all(chunk.as_bytes()).await?;
3747 }
3748 stdin.flush().await?;
3749
3750 let output = child.output().await?;
3751 if !output.status.success() {
3752 return Err(anyhow!(
3753 "command failed with exit code {:?}:\nstdout: {}\nstderr: {}",
3754 output.status.code(),
3755 String::from_utf8_lossy(&output.stdout),
3756 String::from_utf8_lossy(&output.stderr),
3757 ));
3758 }
3759
3760 let stdout = String::from_utf8(output.stdout)?;
3761 Ok(Some(
3762 buffer
3763 .read_with(cx, |buffer, cx| buffer.diff(stdout, cx))
3764 .await,
3765 ))
3766 } else {
3767 Ok(None)
3768 }
3769 }
3770
3771 pub fn definition<T: ToPointUtf16>(
3772 &self,
3773 buffer: &ModelHandle<Buffer>,
3774 position: T,
3775 cx: &mut ModelContext<Self>,
3776 ) -> Task<Result<Vec<LocationLink>>> {
3777 let position = position.to_point_utf16(buffer.read(cx));
3778 self.request_lsp(buffer.clone(), GetDefinition { position }, cx)
3779 }
3780
3781 pub fn type_definition<T: ToPointUtf16>(
3782 &self,
3783 buffer: &ModelHandle<Buffer>,
3784 position: T,
3785 cx: &mut ModelContext<Self>,
3786 ) -> Task<Result<Vec<LocationLink>>> {
3787 let position = position.to_point_utf16(buffer.read(cx));
3788 self.request_lsp(buffer.clone(), GetTypeDefinition { position }, cx)
3789 }
3790
3791 pub fn references<T: ToPointUtf16>(
3792 &self,
3793 buffer: &ModelHandle<Buffer>,
3794 position: T,
3795 cx: &mut ModelContext<Self>,
3796 ) -> Task<Result<Vec<Location>>> {
3797 let position = position.to_point_utf16(buffer.read(cx));
3798 self.request_lsp(buffer.clone(), GetReferences { position }, cx)
3799 }
3800
3801 pub fn document_highlights<T: ToPointUtf16>(
3802 &self,
3803 buffer: &ModelHandle<Buffer>,
3804 position: T,
3805 cx: &mut ModelContext<Self>,
3806 ) -> Task<Result<Vec<DocumentHighlight>>> {
3807 let position = position.to_point_utf16(buffer.read(cx));
3808 self.request_lsp(buffer.clone(), GetDocumentHighlights { position }, cx)
3809 }
3810
3811 pub fn symbols(&self, query: &str, cx: &mut ModelContext<Self>) -> Task<Result<Vec<Symbol>>> {
3812 if self.is_local() {
3813 let mut requests = Vec::new();
3814 for ((worktree_id, _), server_id) in self.language_server_ids.iter() {
3815 let worktree_id = *worktree_id;
3816 if let Some(worktree) = self
3817 .worktree_for_id(worktree_id, cx)
3818 .and_then(|worktree| worktree.read(cx).as_local())
3819 {
3820 if let Some(LanguageServerState::Running {
3821 adapter,
3822 language,
3823 server,
3824 ..
3825 }) = self.language_servers.get(server_id)
3826 {
3827 let adapter = adapter.clone();
3828 let language = language.clone();
3829 let worktree_abs_path = worktree.abs_path().clone();
3830 requests.push(
3831 server
3832 .request::<lsp::request::WorkspaceSymbolRequest>(
3833 lsp::WorkspaceSymbolParams {
3834 query: query.to_string(),
3835 ..Default::default()
3836 },
3837 )
3838 .log_err()
3839 .map(move |response| {
3840 let lsp_symbols = response.flatten().map(|symbol_response| match symbol_response {
3841 lsp::WorkspaceSymbolResponse::Flat(flat_responses) => {
3842 flat_responses.into_iter().map(|lsp_symbol| {
3843 (lsp_symbol.name, lsp_symbol.kind, lsp_symbol.location)
3844 }).collect::<Vec<_>>()
3845 }
3846 lsp::WorkspaceSymbolResponse::Nested(nested_responses) => {
3847 nested_responses.into_iter().filter_map(|lsp_symbol| {
3848 let location = match lsp_symbol.location {
3849 lsp::OneOf::Left(location) => location,
3850 lsp::OneOf::Right(_) => {
3851 error!("Unexpected: client capabilities forbid symbol resolutions in workspace.symbol.resolveSupport");
3852 return None
3853 }
3854 };
3855 Some((lsp_symbol.name, lsp_symbol.kind, location))
3856 }).collect::<Vec<_>>()
3857 }
3858 }).unwrap_or_default();
3859
3860 (
3861 adapter,
3862 language,
3863 worktree_id,
3864 worktree_abs_path,
3865 lsp_symbols,
3866 )
3867 }),
3868 );
3869 }
3870 }
3871 }
3872
3873 cx.spawn_weak(|this, cx| async move {
3874 let responses = futures::future::join_all(requests).await;
3875 let this = if let Some(this) = this.upgrade(&cx) {
3876 this
3877 } else {
3878 return Ok(Default::default());
3879 };
3880 let symbols = this.read_with(&cx, |this, cx| {
3881 let mut symbols = Vec::new();
3882 for (
3883 adapter,
3884 adapter_language,
3885 source_worktree_id,
3886 worktree_abs_path,
3887 lsp_symbols,
3888 ) in responses
3889 {
3890 symbols.extend(lsp_symbols.into_iter().filter_map(
3891 |(symbol_name, symbol_kind, symbol_location)| {
3892 let abs_path = symbol_location.uri.to_file_path().ok()?;
3893 let mut worktree_id = source_worktree_id;
3894 let path;
3895 if let Some((worktree, rel_path)) =
3896 this.find_local_worktree(&abs_path, cx)
3897 {
3898 worktree_id = worktree.read(cx).id();
3899 path = rel_path;
3900 } else {
3901 path = relativize_path(&worktree_abs_path, &abs_path);
3902 }
3903
3904 let project_path = ProjectPath {
3905 worktree_id,
3906 path: path.into(),
3907 };
3908 let signature = this.symbol_signature(&project_path);
3909 let adapter_language = adapter_language.clone();
3910 let language = this
3911 .languages
3912 .language_for_file(&project_path.path, None)
3913 .unwrap_or_else(move |_| adapter_language);
3914 let language_server_name = adapter.name.clone();
3915 Some(async move {
3916 let language = language.await;
3917 let label =
3918 language.label_for_symbol(&symbol_name, symbol_kind).await;
3919
3920 Symbol {
3921 language_server_name,
3922 source_worktree_id,
3923 path: project_path,
3924 label: label.unwrap_or_else(|| {
3925 CodeLabel::plain(symbol_name.clone(), None)
3926 }),
3927 kind: symbol_kind,
3928 name: symbol_name,
3929 range: range_from_lsp(symbol_location.range),
3930 signature,
3931 }
3932 })
3933 },
3934 ));
3935 }
3936 symbols
3937 });
3938 Ok(futures::future::join_all(symbols).await)
3939 })
3940 } else if let Some(project_id) = self.remote_id() {
3941 let request = self.client.request(proto::GetProjectSymbols {
3942 project_id,
3943 query: query.to_string(),
3944 });
3945 cx.spawn_weak(|this, cx| async move {
3946 let response = request.await?;
3947 let mut symbols = Vec::new();
3948 if let Some(this) = this.upgrade(&cx) {
3949 let new_symbols = this.read_with(&cx, |this, _| {
3950 response
3951 .symbols
3952 .into_iter()
3953 .map(|symbol| this.deserialize_symbol(symbol))
3954 .collect::<Vec<_>>()
3955 });
3956 symbols = futures::future::join_all(new_symbols)
3957 .await
3958 .into_iter()
3959 .filter_map(|symbol| symbol.log_err())
3960 .collect::<Vec<_>>();
3961 }
3962 Ok(symbols)
3963 })
3964 } else {
3965 Task::ready(Ok(Default::default()))
3966 }
3967 }
3968
3969 pub fn open_buffer_for_symbol(
3970 &mut self,
3971 symbol: &Symbol,
3972 cx: &mut ModelContext<Self>,
3973 ) -> Task<Result<ModelHandle<Buffer>>> {
3974 if self.is_local() {
3975 let language_server_id = if let Some(id) = self.language_server_ids.get(&(
3976 symbol.source_worktree_id,
3977 symbol.language_server_name.clone(),
3978 )) {
3979 *id
3980 } else {
3981 return Task::ready(Err(anyhow!(
3982 "language server for worktree and language not found"
3983 )));
3984 };
3985
3986 let worktree_abs_path = if let Some(worktree_abs_path) = self
3987 .worktree_for_id(symbol.path.worktree_id, cx)
3988 .and_then(|worktree| worktree.read(cx).as_local())
3989 .map(|local_worktree| local_worktree.abs_path())
3990 {
3991 worktree_abs_path
3992 } else {
3993 return Task::ready(Err(anyhow!("worktree not found for symbol")));
3994 };
3995 let symbol_abs_path = worktree_abs_path.join(&symbol.path.path);
3996 let symbol_uri = if let Ok(uri) = lsp::Url::from_file_path(symbol_abs_path) {
3997 uri
3998 } else {
3999 return Task::ready(Err(anyhow!("invalid symbol path")));
4000 };
4001
4002 self.open_local_buffer_via_lsp(
4003 symbol_uri,
4004 language_server_id,
4005 symbol.language_server_name.clone(),
4006 cx,
4007 )
4008 } else if let Some(project_id) = self.remote_id() {
4009 let request = self.client.request(proto::OpenBufferForSymbol {
4010 project_id,
4011 symbol: Some(serialize_symbol(symbol)),
4012 });
4013 cx.spawn(|this, mut cx| async move {
4014 let response = request.await?;
4015 this.update(&mut cx, |this, cx| {
4016 this.wait_for_remote_buffer(response.buffer_id, cx)
4017 })
4018 .await
4019 })
4020 } else {
4021 Task::ready(Err(anyhow!("project does not have a remote id")))
4022 }
4023 }
4024
4025 pub fn hover<T: ToPointUtf16>(
4026 &self,
4027 buffer: &ModelHandle<Buffer>,
4028 position: T,
4029 cx: &mut ModelContext<Self>,
4030 ) -> Task<Result<Option<Hover>>> {
4031 let position = position.to_point_utf16(buffer.read(cx));
4032 self.request_lsp(buffer.clone(), GetHover { position }, cx)
4033 }
4034
4035 pub fn completions<T: ToPointUtf16>(
4036 &self,
4037 buffer: &ModelHandle<Buffer>,
4038 position: T,
4039 cx: &mut ModelContext<Self>,
4040 ) -> Task<Result<Vec<Completion>>> {
4041 let position = position.to_point_utf16(buffer.read(cx));
4042 self.request_lsp(buffer.clone(), GetCompletions { position }, cx)
4043 }
4044
4045 pub fn apply_additional_edits_for_completion(
4046 &self,
4047 buffer_handle: ModelHandle<Buffer>,
4048 completion: Completion,
4049 push_to_history: bool,
4050 cx: &mut ModelContext<Self>,
4051 ) -> Task<Result<Option<Transaction>>> {
4052 let buffer = buffer_handle.read(cx);
4053 let buffer_id = buffer.remote_id();
4054
4055 if self.is_local() {
4056 let lang_server = match self.primary_language_servers_for_buffer(buffer, cx) {
4057 Some((_, server)) => server.clone(),
4058 _ => return Task::ready(Ok(Default::default())),
4059 };
4060
4061 cx.spawn(|this, mut cx| async move {
4062 let resolved_completion = lang_server
4063 .request::<lsp::request::ResolveCompletionItem>(completion.lsp_completion)
4064 .await?;
4065
4066 if let Some(edits) = resolved_completion.additional_text_edits {
4067 let edits = this
4068 .update(&mut cx, |this, cx| {
4069 this.edits_from_lsp(
4070 &buffer_handle,
4071 edits,
4072 lang_server.server_id(),
4073 None,
4074 cx,
4075 )
4076 })
4077 .await?;
4078
4079 buffer_handle.update(&mut cx, |buffer, cx| {
4080 buffer.finalize_last_transaction();
4081 buffer.start_transaction();
4082
4083 for (range, text) in edits {
4084 let primary = &completion.old_range;
4085 let start_within = primary.start.cmp(&range.start, buffer).is_le()
4086 && primary.end.cmp(&range.start, buffer).is_ge();
4087 let end_within = range.start.cmp(&primary.end, buffer).is_le()
4088 && range.end.cmp(&primary.end, buffer).is_ge();
4089
4090 //Skip additional edits which overlap with the primary completion edit
4091 //https://github.com/zed-industries/zed/pull/1871
4092 if !start_within && !end_within {
4093 buffer.edit([(range, text)], None, cx);
4094 }
4095 }
4096
4097 let transaction = if buffer.end_transaction(cx).is_some() {
4098 let transaction = buffer.finalize_last_transaction().unwrap().clone();
4099 if !push_to_history {
4100 buffer.forget_transaction(transaction.id);
4101 }
4102 Some(transaction)
4103 } else {
4104 None
4105 };
4106 Ok(transaction)
4107 })
4108 } else {
4109 Ok(None)
4110 }
4111 })
4112 } else if let Some(project_id) = self.remote_id() {
4113 let client = self.client.clone();
4114 cx.spawn(|_, mut cx| async move {
4115 let response = client
4116 .request(proto::ApplyCompletionAdditionalEdits {
4117 project_id,
4118 buffer_id,
4119 completion: Some(language::proto::serialize_completion(&completion)),
4120 })
4121 .await?;
4122
4123 if let Some(transaction) = response.transaction {
4124 let transaction = language::proto::deserialize_transaction(transaction)?;
4125 buffer_handle
4126 .update(&mut cx, |buffer, _| {
4127 buffer.wait_for_edits(transaction.edit_ids.iter().copied())
4128 })
4129 .await?;
4130 if push_to_history {
4131 buffer_handle.update(&mut cx, |buffer, _| {
4132 buffer.push_transaction(transaction.clone(), Instant::now());
4133 });
4134 }
4135 Ok(Some(transaction))
4136 } else {
4137 Ok(None)
4138 }
4139 })
4140 } else {
4141 Task::ready(Err(anyhow!("project does not have a remote id")))
4142 }
4143 }
4144
4145 pub fn code_actions<T: Clone + ToOffset>(
4146 &self,
4147 buffer_handle: &ModelHandle<Buffer>,
4148 range: Range<T>,
4149 cx: &mut ModelContext<Self>,
4150 ) -> Task<Result<Vec<CodeAction>>> {
4151 let buffer = buffer_handle.read(cx);
4152 let range = buffer.anchor_before(range.start)..buffer.anchor_before(range.end);
4153 self.request_lsp(buffer_handle.clone(), GetCodeActions { range }, cx)
4154 }
4155
4156 pub fn apply_code_action(
4157 &self,
4158 buffer_handle: ModelHandle<Buffer>,
4159 mut action: CodeAction,
4160 push_to_history: bool,
4161 cx: &mut ModelContext<Self>,
4162 ) -> Task<Result<ProjectTransaction>> {
4163 if self.is_local() {
4164 let buffer = buffer_handle.read(cx);
4165 let (lsp_adapter, lang_server) = if let Some((adapter, server)) =
4166 self.language_server_for_buffer(buffer, action.server_id, cx)
4167 {
4168 (adapter.clone(), server.clone())
4169 } else {
4170 return Task::ready(Ok(Default::default()));
4171 };
4172 let range = action.range.to_point_utf16(buffer);
4173
4174 cx.spawn(|this, mut cx| async move {
4175 if let Some(lsp_range) = action
4176 .lsp_action
4177 .data
4178 .as_mut()
4179 .and_then(|d| d.get_mut("codeActionParams"))
4180 .and_then(|d| d.get_mut("range"))
4181 {
4182 *lsp_range = serde_json::to_value(&range_to_lsp(range)).unwrap();
4183 action.lsp_action = lang_server
4184 .request::<lsp::request::CodeActionResolveRequest>(action.lsp_action)
4185 .await?;
4186 } else {
4187 let actions = this
4188 .update(&mut cx, |this, cx| {
4189 this.code_actions(&buffer_handle, action.range, cx)
4190 })
4191 .await?;
4192 action.lsp_action = actions
4193 .into_iter()
4194 .find(|a| a.lsp_action.title == action.lsp_action.title)
4195 .ok_or_else(|| anyhow!("code action is outdated"))?
4196 .lsp_action;
4197 }
4198
4199 if let Some(edit) = action.lsp_action.edit {
4200 if edit.changes.is_some() || edit.document_changes.is_some() {
4201 return Self::deserialize_workspace_edit(
4202 this,
4203 edit,
4204 push_to_history,
4205 lsp_adapter.clone(),
4206 lang_server.clone(),
4207 &mut cx,
4208 )
4209 .await;
4210 }
4211 }
4212
4213 if let Some(command) = action.lsp_action.command {
4214 this.update(&mut cx, |this, _| {
4215 this.last_workspace_edits_by_language_server
4216 .remove(&lang_server.server_id());
4217 });
4218 lang_server
4219 .request::<lsp::request::ExecuteCommand>(lsp::ExecuteCommandParams {
4220 command: command.command,
4221 arguments: command.arguments.unwrap_or_default(),
4222 ..Default::default()
4223 })
4224 .await?;
4225 return Ok(this.update(&mut cx, |this, _| {
4226 this.last_workspace_edits_by_language_server
4227 .remove(&lang_server.server_id())
4228 .unwrap_or_default()
4229 }));
4230 }
4231
4232 Ok(ProjectTransaction::default())
4233 })
4234 } else if let Some(project_id) = self.remote_id() {
4235 let client = self.client.clone();
4236 let request = proto::ApplyCodeAction {
4237 project_id,
4238 buffer_id: buffer_handle.read(cx).remote_id(),
4239 action: Some(language::proto::serialize_code_action(&action)),
4240 };
4241 cx.spawn(|this, mut cx| async move {
4242 let response = client
4243 .request(request)
4244 .await?
4245 .transaction
4246 .ok_or_else(|| anyhow!("missing transaction"))?;
4247 this.update(&mut cx, |this, cx| {
4248 this.deserialize_project_transaction(response, push_to_history, cx)
4249 })
4250 .await
4251 })
4252 } else {
4253 Task::ready(Err(anyhow!("project does not have a remote id")))
4254 }
4255 }
4256
4257 fn apply_on_type_formatting(
4258 &self,
4259 buffer: ModelHandle<Buffer>,
4260 position: Anchor,
4261 trigger: String,
4262 cx: &mut ModelContext<Self>,
4263 ) -> Task<Result<Option<Transaction>>> {
4264 if self.is_local() {
4265 cx.spawn(|this, mut cx| async move {
4266 // Do not allow multiple concurrent formatting requests for the
4267 // same buffer.
4268 this.update(&mut cx, |this, cx| {
4269 this.buffers_being_formatted
4270 .insert(buffer.read(cx).remote_id())
4271 });
4272
4273 let _cleanup = defer({
4274 let this = this.clone();
4275 let mut cx = cx.clone();
4276 let closure_buffer = buffer.clone();
4277 move || {
4278 this.update(&mut cx, |this, cx| {
4279 this.buffers_being_formatted
4280 .remove(&closure_buffer.read(cx).remote_id());
4281 });
4282 }
4283 });
4284
4285 buffer
4286 .update(&mut cx, |buffer, _| {
4287 buffer.wait_for_edits(Some(position.timestamp))
4288 })
4289 .await?;
4290 this.update(&mut cx, |this, cx| {
4291 let position = position.to_point_utf16(buffer.read(cx));
4292 this.on_type_format(buffer, position, trigger, false, cx)
4293 })
4294 .await
4295 })
4296 } else if let Some(project_id) = self.remote_id() {
4297 let client = self.client.clone();
4298 let request = proto::OnTypeFormatting {
4299 project_id,
4300 buffer_id: buffer.read(cx).remote_id(),
4301 position: Some(serialize_anchor(&position)),
4302 trigger,
4303 version: serialize_version(&buffer.read(cx).version()),
4304 };
4305 cx.spawn(|_, _| async move {
4306 client
4307 .request(request)
4308 .await?
4309 .transaction
4310 .map(language::proto::deserialize_transaction)
4311 .transpose()
4312 })
4313 } else {
4314 Task::ready(Err(anyhow!("project does not have a remote id")))
4315 }
4316 }
4317
4318 async fn deserialize_edits(
4319 this: ModelHandle<Self>,
4320 buffer_to_edit: ModelHandle<Buffer>,
4321 edits: Vec<lsp::TextEdit>,
4322 push_to_history: bool,
4323 _: Arc<CachedLspAdapter>,
4324 language_server: Arc<LanguageServer>,
4325 cx: &mut AsyncAppContext,
4326 ) -> Result<Option<Transaction>> {
4327 let edits = this
4328 .update(cx, |this, cx| {
4329 this.edits_from_lsp(
4330 &buffer_to_edit,
4331 edits,
4332 language_server.server_id(),
4333 None,
4334 cx,
4335 )
4336 })
4337 .await?;
4338
4339 let transaction = buffer_to_edit.update(cx, |buffer, cx| {
4340 buffer.finalize_last_transaction();
4341 buffer.start_transaction();
4342 for (range, text) in edits {
4343 buffer.edit([(range, text)], None, cx);
4344 }
4345
4346 if buffer.end_transaction(cx).is_some() {
4347 let transaction = buffer.finalize_last_transaction().unwrap().clone();
4348 if !push_to_history {
4349 buffer.forget_transaction(transaction.id);
4350 }
4351 Some(transaction)
4352 } else {
4353 None
4354 }
4355 });
4356
4357 Ok(transaction)
4358 }
4359
4360 async fn deserialize_workspace_edit(
4361 this: ModelHandle<Self>,
4362 edit: lsp::WorkspaceEdit,
4363 push_to_history: bool,
4364 lsp_adapter: Arc<CachedLspAdapter>,
4365 language_server: Arc<LanguageServer>,
4366 cx: &mut AsyncAppContext,
4367 ) -> Result<ProjectTransaction> {
4368 let fs = this.read_with(cx, |this, _| this.fs.clone());
4369 let mut operations = Vec::new();
4370 if let Some(document_changes) = edit.document_changes {
4371 match document_changes {
4372 lsp::DocumentChanges::Edits(edits) => {
4373 operations.extend(edits.into_iter().map(lsp::DocumentChangeOperation::Edit))
4374 }
4375 lsp::DocumentChanges::Operations(ops) => operations = ops,
4376 }
4377 } else if let Some(changes) = edit.changes {
4378 operations.extend(changes.into_iter().map(|(uri, edits)| {
4379 lsp::DocumentChangeOperation::Edit(lsp::TextDocumentEdit {
4380 text_document: lsp::OptionalVersionedTextDocumentIdentifier {
4381 uri,
4382 version: None,
4383 },
4384 edits: edits.into_iter().map(lsp::OneOf::Left).collect(),
4385 })
4386 }));
4387 }
4388
4389 let mut project_transaction = ProjectTransaction::default();
4390 for operation in operations {
4391 match operation {
4392 lsp::DocumentChangeOperation::Op(lsp::ResourceOp::Create(op)) => {
4393 let abs_path = op
4394 .uri
4395 .to_file_path()
4396 .map_err(|_| anyhow!("can't convert URI to path"))?;
4397
4398 if let Some(parent_path) = abs_path.parent() {
4399 fs.create_dir(parent_path).await?;
4400 }
4401 if abs_path.ends_with("/") {
4402 fs.create_dir(&abs_path).await?;
4403 } else {
4404 fs.create_file(&abs_path, op.options.map(Into::into).unwrap_or_default())
4405 .await?;
4406 }
4407 }
4408
4409 lsp::DocumentChangeOperation::Op(lsp::ResourceOp::Rename(op)) => {
4410 let source_abs_path = op
4411 .old_uri
4412 .to_file_path()
4413 .map_err(|_| anyhow!("can't convert URI to path"))?;
4414 let target_abs_path = op
4415 .new_uri
4416 .to_file_path()
4417 .map_err(|_| anyhow!("can't convert URI to path"))?;
4418 fs.rename(
4419 &source_abs_path,
4420 &target_abs_path,
4421 op.options.map(Into::into).unwrap_or_default(),
4422 )
4423 .await?;
4424 }
4425
4426 lsp::DocumentChangeOperation::Op(lsp::ResourceOp::Delete(op)) => {
4427 let abs_path = op
4428 .uri
4429 .to_file_path()
4430 .map_err(|_| anyhow!("can't convert URI to path"))?;
4431 let options = op.options.map(Into::into).unwrap_or_default();
4432 if abs_path.ends_with("/") {
4433 fs.remove_dir(&abs_path, options).await?;
4434 } else {
4435 fs.remove_file(&abs_path, options).await?;
4436 }
4437 }
4438
4439 lsp::DocumentChangeOperation::Edit(op) => {
4440 let buffer_to_edit = this
4441 .update(cx, |this, cx| {
4442 this.open_local_buffer_via_lsp(
4443 op.text_document.uri,
4444 language_server.server_id(),
4445 lsp_adapter.name.clone(),
4446 cx,
4447 )
4448 })
4449 .await?;
4450
4451 let edits = this
4452 .update(cx, |this, cx| {
4453 let edits = op.edits.into_iter().map(|edit| match edit {
4454 lsp::OneOf::Left(edit) => edit,
4455 lsp::OneOf::Right(edit) => edit.text_edit,
4456 });
4457 this.edits_from_lsp(
4458 &buffer_to_edit,
4459 edits,
4460 language_server.server_id(),
4461 op.text_document.version,
4462 cx,
4463 )
4464 })
4465 .await?;
4466
4467 let transaction = buffer_to_edit.update(cx, |buffer, cx| {
4468 buffer.finalize_last_transaction();
4469 buffer.start_transaction();
4470 for (range, text) in edits {
4471 buffer.edit([(range, text)], None, cx);
4472 }
4473 let transaction = if buffer.end_transaction(cx).is_some() {
4474 let transaction = buffer.finalize_last_transaction().unwrap().clone();
4475 if !push_to_history {
4476 buffer.forget_transaction(transaction.id);
4477 }
4478 Some(transaction)
4479 } else {
4480 None
4481 };
4482
4483 transaction
4484 });
4485 if let Some(transaction) = transaction {
4486 project_transaction.0.insert(buffer_to_edit, transaction);
4487 }
4488 }
4489 }
4490 }
4491
4492 Ok(project_transaction)
4493 }
4494
4495 pub fn prepare_rename<T: ToPointUtf16>(
4496 &self,
4497 buffer: ModelHandle<Buffer>,
4498 position: T,
4499 cx: &mut ModelContext<Self>,
4500 ) -> Task<Result<Option<Range<Anchor>>>> {
4501 let position = position.to_point_utf16(buffer.read(cx));
4502 self.request_lsp(buffer, PrepareRename { position }, cx)
4503 }
4504
4505 pub fn perform_rename<T: ToPointUtf16>(
4506 &self,
4507 buffer: ModelHandle<Buffer>,
4508 position: T,
4509 new_name: String,
4510 push_to_history: bool,
4511 cx: &mut ModelContext<Self>,
4512 ) -> Task<Result<ProjectTransaction>> {
4513 let position = position.to_point_utf16(buffer.read(cx));
4514 self.request_lsp(
4515 buffer,
4516 PerformRename {
4517 position,
4518 new_name,
4519 push_to_history,
4520 },
4521 cx,
4522 )
4523 }
4524
4525 pub fn on_type_format<T: ToPointUtf16>(
4526 &self,
4527 buffer: ModelHandle<Buffer>,
4528 position: T,
4529 trigger: String,
4530 push_to_history: bool,
4531 cx: &mut ModelContext<Self>,
4532 ) -> Task<Result<Option<Transaction>>> {
4533 let (position, tab_size) = buffer.read_with(cx, |buffer, cx| {
4534 let position = position.to_point_utf16(buffer);
4535 (
4536 position,
4537 language_settings(buffer.language_at(position).as_ref(), buffer.file(), cx)
4538 .tab_size,
4539 )
4540 });
4541 self.request_lsp(
4542 buffer.clone(),
4543 OnTypeFormatting {
4544 position,
4545 trigger,
4546 options: lsp_command::lsp_formatting_options(tab_size.get()).into(),
4547 push_to_history,
4548 },
4549 cx,
4550 )
4551 }
4552
4553 #[allow(clippy::type_complexity)]
4554 pub fn search(
4555 &self,
4556 query: SearchQuery,
4557 cx: &mut ModelContext<Self>,
4558 ) -> Task<Result<HashMap<ModelHandle<Buffer>, Vec<Range<Anchor>>>>> {
4559 if self.is_local() {
4560 let snapshots = self
4561 .visible_worktrees(cx)
4562 .filter_map(|tree| {
4563 let tree = tree.read(cx).as_local()?;
4564 Some(tree.snapshot())
4565 })
4566 .collect::<Vec<_>>();
4567
4568 let background = cx.background().clone();
4569 let path_count: usize = snapshots.iter().map(|s| s.visible_file_count()).sum();
4570 if path_count == 0 {
4571 return Task::ready(Ok(Default::default()));
4572 }
4573 let workers = background.num_cpus().min(path_count);
4574 let (matching_paths_tx, mut matching_paths_rx) = smol::channel::bounded(1024);
4575 cx.background()
4576 .spawn({
4577 let fs = self.fs.clone();
4578 let background = cx.background().clone();
4579 let query = query.clone();
4580 async move {
4581 let fs = &fs;
4582 let query = &query;
4583 let matching_paths_tx = &matching_paths_tx;
4584 let paths_per_worker = (path_count + workers - 1) / workers;
4585 let snapshots = &snapshots;
4586 background
4587 .scoped(|scope| {
4588 for worker_ix in 0..workers {
4589 let worker_start_ix = worker_ix * paths_per_worker;
4590 let worker_end_ix = worker_start_ix + paths_per_worker;
4591 scope.spawn(async move {
4592 let mut snapshot_start_ix = 0;
4593 let mut abs_path = PathBuf::new();
4594 for snapshot in snapshots {
4595 let snapshot_end_ix =
4596 snapshot_start_ix + snapshot.visible_file_count();
4597 if worker_end_ix <= snapshot_start_ix {
4598 break;
4599 } else if worker_start_ix > snapshot_end_ix {
4600 snapshot_start_ix = snapshot_end_ix;
4601 continue;
4602 } else {
4603 let start_in_snapshot = worker_start_ix
4604 .saturating_sub(snapshot_start_ix);
4605 let end_in_snapshot =
4606 cmp::min(worker_end_ix, snapshot_end_ix)
4607 - snapshot_start_ix;
4608
4609 for entry in snapshot
4610 .files(false, start_in_snapshot)
4611 .take(end_in_snapshot - start_in_snapshot)
4612 {
4613 if matching_paths_tx.is_closed() {
4614 break;
4615 }
4616 let matches = if query
4617 .file_matches(Some(&entry.path))
4618 {
4619 abs_path.clear();
4620 abs_path.push(&snapshot.abs_path());
4621 abs_path.push(&entry.path);
4622 if let Some(file) =
4623 fs.open_sync(&abs_path).await.log_err()
4624 {
4625 query.detect(file).unwrap_or(false)
4626 } else {
4627 false
4628 }
4629 } else {
4630 false
4631 };
4632
4633 if matches {
4634 let project_path =
4635 (snapshot.id(), entry.path.clone());
4636 if matching_paths_tx
4637 .send(project_path)
4638 .await
4639 .is_err()
4640 {
4641 break;
4642 }
4643 }
4644 }
4645
4646 snapshot_start_ix = snapshot_end_ix;
4647 }
4648 }
4649 });
4650 }
4651 })
4652 .await;
4653 }
4654 })
4655 .detach();
4656
4657 let (buffers_tx, buffers_rx) = smol::channel::bounded(1024);
4658 let open_buffers = self
4659 .opened_buffers
4660 .values()
4661 .filter_map(|b| b.upgrade(cx))
4662 .collect::<HashSet<_>>();
4663 cx.spawn(|this, cx| async move {
4664 for buffer in &open_buffers {
4665 let snapshot = buffer.read_with(&cx, |buffer, _| buffer.snapshot());
4666 buffers_tx.send((buffer.clone(), snapshot)).await?;
4667 }
4668
4669 let open_buffers = Rc::new(RefCell::new(open_buffers));
4670 while let Some(project_path) = matching_paths_rx.next().await {
4671 if buffers_tx.is_closed() {
4672 break;
4673 }
4674
4675 let this = this.clone();
4676 let open_buffers = open_buffers.clone();
4677 let buffers_tx = buffers_tx.clone();
4678 cx.spawn(|mut cx| async move {
4679 if let Some(buffer) = this
4680 .update(&mut cx, |this, cx| this.open_buffer(project_path, cx))
4681 .await
4682 .log_err()
4683 {
4684 if open_buffers.borrow_mut().insert(buffer.clone()) {
4685 let snapshot = buffer.read_with(&cx, |buffer, _| buffer.snapshot());
4686 buffers_tx.send((buffer, snapshot)).await?;
4687 }
4688 }
4689
4690 Ok::<_, anyhow::Error>(())
4691 })
4692 .detach();
4693 }
4694
4695 Ok::<_, anyhow::Error>(())
4696 })
4697 .detach_and_log_err(cx);
4698
4699 let background = cx.background().clone();
4700 cx.background().spawn(async move {
4701 let query = &query;
4702 let mut matched_buffers = Vec::new();
4703 for _ in 0..workers {
4704 matched_buffers.push(HashMap::default());
4705 }
4706 background
4707 .scoped(|scope| {
4708 for worker_matched_buffers in matched_buffers.iter_mut() {
4709 let mut buffers_rx = buffers_rx.clone();
4710 scope.spawn(async move {
4711 while let Some((buffer, snapshot)) = buffers_rx.next().await {
4712 let buffer_matches = if query.file_matches(
4713 snapshot.file().map(|file| file.path().as_ref()),
4714 ) {
4715 query
4716 .search(snapshot.as_rope())
4717 .await
4718 .iter()
4719 .map(|range| {
4720 snapshot.anchor_before(range.start)
4721 ..snapshot.anchor_after(range.end)
4722 })
4723 .collect()
4724 } else {
4725 Vec::new()
4726 };
4727 if !buffer_matches.is_empty() {
4728 worker_matched_buffers
4729 .insert(buffer.clone(), buffer_matches);
4730 }
4731 }
4732 });
4733 }
4734 })
4735 .await;
4736 Ok(matched_buffers.into_iter().flatten().collect())
4737 })
4738 } else if let Some(project_id) = self.remote_id() {
4739 let request = self.client.request(query.to_proto(project_id));
4740 cx.spawn(|this, mut cx| async move {
4741 let response = request.await?;
4742 let mut result = HashMap::default();
4743 for location in response.locations {
4744 let target_buffer = this
4745 .update(&mut cx, |this, cx| {
4746 this.wait_for_remote_buffer(location.buffer_id, cx)
4747 })
4748 .await?;
4749 let start = location
4750 .start
4751 .and_then(deserialize_anchor)
4752 .ok_or_else(|| anyhow!("missing target start"))?;
4753 let end = location
4754 .end
4755 .and_then(deserialize_anchor)
4756 .ok_or_else(|| anyhow!("missing target end"))?;
4757 result
4758 .entry(target_buffer)
4759 .or_insert(Vec::new())
4760 .push(start..end)
4761 }
4762 Ok(result)
4763 })
4764 } else {
4765 Task::ready(Ok(Default::default()))
4766 }
4767 }
4768
4769 // TODO: Wire this up to allow selecting a server?
4770 fn request_lsp<R: LspCommand>(
4771 &self,
4772 buffer_handle: ModelHandle<Buffer>,
4773 request: R,
4774 cx: &mut ModelContext<Self>,
4775 ) -> Task<Result<R::Response>>
4776 where
4777 <R::LspRequest as lsp::request::Request>::Result: Send,
4778 {
4779 let buffer = buffer_handle.read(cx);
4780 if self.is_local() {
4781 let file = File::from_dyn(buffer.file()).and_then(File::as_local);
4782 if let Some((file, language_server)) = file.zip(
4783 self.primary_language_servers_for_buffer(buffer, cx)
4784 .map(|(_, server)| server.clone()),
4785 ) {
4786 let lsp_params = request.to_lsp(&file.abs_path(cx), buffer, &language_server, cx);
4787 return cx.spawn(|this, cx| async move {
4788 if !request.check_capabilities(language_server.capabilities()) {
4789 return Ok(Default::default());
4790 }
4791
4792 let response = language_server
4793 .request::<R::LspRequest>(lsp_params)
4794 .await
4795 .context("lsp request failed")?;
4796 request
4797 .response_from_lsp(
4798 response,
4799 this,
4800 buffer_handle,
4801 language_server.server_id(),
4802 cx,
4803 )
4804 .await
4805 });
4806 }
4807 } else if let Some(project_id) = self.remote_id() {
4808 let rpc = self.client.clone();
4809 let message = request.to_proto(project_id, buffer);
4810 return cx.spawn_weak(|this, cx| async move {
4811 // Ensure the project is still alive by the time the task
4812 // is scheduled.
4813 this.upgrade(&cx)
4814 .ok_or_else(|| anyhow!("project dropped"))?;
4815
4816 let response = rpc.request(message).await?;
4817
4818 let this = this
4819 .upgrade(&cx)
4820 .ok_or_else(|| anyhow!("project dropped"))?;
4821 if this.read_with(&cx, |this, _| this.is_read_only()) {
4822 Err(anyhow!("disconnected before completing request"))
4823 } else {
4824 request
4825 .response_from_proto(response, this, buffer_handle, cx)
4826 .await
4827 }
4828 });
4829 }
4830 Task::ready(Ok(Default::default()))
4831 }
4832
4833 pub fn find_or_create_local_worktree(
4834 &mut self,
4835 abs_path: impl AsRef<Path>,
4836 visible: bool,
4837 cx: &mut ModelContext<Self>,
4838 ) -> Task<Result<(ModelHandle<Worktree>, PathBuf)>> {
4839 let abs_path = abs_path.as_ref();
4840 if let Some((tree, relative_path)) = self.find_local_worktree(abs_path, cx) {
4841 Task::ready(Ok((tree, relative_path)))
4842 } else {
4843 let worktree = self.create_local_worktree(abs_path, visible, cx);
4844 cx.foreground()
4845 .spawn(async move { Ok((worktree.await?, PathBuf::new())) })
4846 }
4847 }
4848
4849 pub fn find_local_worktree(
4850 &self,
4851 abs_path: &Path,
4852 cx: &AppContext,
4853 ) -> Option<(ModelHandle<Worktree>, PathBuf)> {
4854 for tree in &self.worktrees {
4855 if let Some(tree) = tree.upgrade(cx) {
4856 if let Some(relative_path) = tree
4857 .read(cx)
4858 .as_local()
4859 .and_then(|t| abs_path.strip_prefix(t.abs_path()).ok())
4860 {
4861 return Some((tree.clone(), relative_path.into()));
4862 }
4863 }
4864 }
4865 None
4866 }
4867
4868 pub fn is_shared(&self) -> bool {
4869 match &self.client_state {
4870 Some(ProjectClientState::Local { .. }) => true,
4871 _ => false,
4872 }
4873 }
4874
4875 fn create_local_worktree(
4876 &mut self,
4877 abs_path: impl AsRef<Path>,
4878 visible: bool,
4879 cx: &mut ModelContext<Self>,
4880 ) -> Task<Result<ModelHandle<Worktree>>> {
4881 let fs = self.fs.clone();
4882 let client = self.client.clone();
4883 let next_entry_id = self.next_entry_id.clone();
4884 let path: Arc<Path> = abs_path.as_ref().into();
4885 let task = self
4886 .loading_local_worktrees
4887 .entry(path.clone())
4888 .or_insert_with(|| {
4889 cx.spawn(|project, mut cx| {
4890 async move {
4891 let worktree = Worktree::local(
4892 client.clone(),
4893 path.clone(),
4894 visible,
4895 fs,
4896 next_entry_id,
4897 &mut cx,
4898 )
4899 .await;
4900
4901 project.update(&mut cx, |project, _| {
4902 project.loading_local_worktrees.remove(&path);
4903 });
4904
4905 let worktree = worktree?;
4906 project.update(&mut cx, |project, cx| project.add_worktree(&worktree, cx));
4907 Ok(worktree)
4908 }
4909 .map_err(Arc::new)
4910 })
4911 .shared()
4912 })
4913 .clone();
4914 cx.foreground().spawn(async move {
4915 match task.await {
4916 Ok(worktree) => Ok(worktree),
4917 Err(err) => Err(anyhow!("{}", err)),
4918 }
4919 })
4920 }
4921
4922 pub fn remove_worktree(&mut self, id_to_remove: WorktreeId, cx: &mut ModelContext<Self>) {
4923 self.worktrees.retain(|worktree| {
4924 if let Some(worktree) = worktree.upgrade(cx) {
4925 let id = worktree.read(cx).id();
4926 if id == id_to_remove {
4927 cx.emit(Event::WorktreeRemoved(id));
4928 false
4929 } else {
4930 true
4931 }
4932 } else {
4933 false
4934 }
4935 });
4936 self.metadata_changed(cx);
4937 }
4938
4939 fn add_worktree(&mut self, worktree: &ModelHandle<Worktree>, cx: &mut ModelContext<Self>) {
4940 cx.observe(worktree, |_, _, cx| cx.notify()).detach();
4941 if worktree.read(cx).is_local() {
4942 cx.subscribe(worktree, |this, worktree, event, cx| match event {
4943 worktree::Event::UpdatedEntries(changes) => {
4944 this.update_local_worktree_buffers(&worktree, changes, cx);
4945 this.update_local_worktree_language_servers(&worktree, changes, cx);
4946 this.update_local_worktree_settings(&worktree, changes, cx);
4947 }
4948 worktree::Event::UpdatedGitRepositories(updated_repos) => {
4949 this.update_local_worktree_buffers_git_repos(worktree, updated_repos, cx)
4950 }
4951 })
4952 .detach();
4953 }
4954
4955 let push_strong_handle = {
4956 let worktree = worktree.read(cx);
4957 self.is_shared() || worktree.is_visible() || worktree.is_remote()
4958 };
4959 if push_strong_handle {
4960 self.worktrees
4961 .push(WorktreeHandle::Strong(worktree.clone()));
4962 } else {
4963 self.worktrees
4964 .push(WorktreeHandle::Weak(worktree.downgrade()));
4965 }
4966
4967 let handle_id = worktree.id();
4968 cx.observe_release(worktree, move |this, worktree, cx| {
4969 let _ = this.remove_worktree(worktree.id(), cx);
4970 cx.update_global::<SettingsStore, _, _>(|store, cx| {
4971 store.clear_local_settings(handle_id, cx).log_err()
4972 });
4973 })
4974 .detach();
4975
4976 cx.emit(Event::WorktreeAdded);
4977 self.metadata_changed(cx);
4978 }
4979
4980 fn update_local_worktree_buffers(
4981 &mut self,
4982 worktree_handle: &ModelHandle<Worktree>,
4983 changes: &[(Arc<Path>, ProjectEntryId, PathChange)],
4984 cx: &mut ModelContext<Self>,
4985 ) {
4986 let snapshot = worktree_handle.read(cx).snapshot();
4987
4988 let mut renamed_buffers = Vec::new();
4989 for (path, entry_id, _) in changes {
4990 let worktree_id = worktree_handle.read(cx).id();
4991 let project_path = ProjectPath {
4992 worktree_id,
4993 path: path.clone(),
4994 };
4995
4996 let buffer_id = match self.local_buffer_ids_by_entry_id.get(entry_id) {
4997 Some(&buffer_id) => buffer_id,
4998 None => match self.local_buffer_ids_by_path.get(&project_path) {
4999 Some(&buffer_id) => buffer_id,
5000 None => continue,
5001 },
5002 };
5003
5004 let open_buffer = self.opened_buffers.get(&buffer_id);
5005 let buffer = if let Some(buffer) = open_buffer.and_then(|buffer| buffer.upgrade(cx)) {
5006 buffer
5007 } else {
5008 self.opened_buffers.remove(&buffer_id);
5009 self.local_buffer_ids_by_path.remove(&project_path);
5010 self.local_buffer_ids_by_entry_id.remove(entry_id);
5011 continue;
5012 };
5013
5014 buffer.update(cx, |buffer, cx| {
5015 if let Some(old_file) = File::from_dyn(buffer.file()) {
5016 if old_file.worktree != *worktree_handle {
5017 return;
5018 }
5019
5020 let new_file = if let Some(entry) = snapshot.entry_for_id(old_file.entry_id) {
5021 File {
5022 is_local: true,
5023 entry_id: entry.id,
5024 mtime: entry.mtime,
5025 path: entry.path.clone(),
5026 worktree: worktree_handle.clone(),
5027 is_deleted: false,
5028 }
5029 } else if let Some(entry) = snapshot.entry_for_path(old_file.path().as_ref()) {
5030 File {
5031 is_local: true,
5032 entry_id: entry.id,
5033 mtime: entry.mtime,
5034 path: entry.path.clone(),
5035 worktree: worktree_handle.clone(),
5036 is_deleted: false,
5037 }
5038 } else {
5039 File {
5040 is_local: true,
5041 entry_id: old_file.entry_id,
5042 path: old_file.path().clone(),
5043 mtime: old_file.mtime(),
5044 worktree: worktree_handle.clone(),
5045 is_deleted: true,
5046 }
5047 };
5048
5049 let old_path = old_file.abs_path(cx);
5050 if new_file.abs_path(cx) != old_path {
5051 renamed_buffers.push((cx.handle(), old_file.clone()));
5052 self.local_buffer_ids_by_path.remove(&project_path);
5053 self.local_buffer_ids_by_path.insert(
5054 ProjectPath {
5055 worktree_id,
5056 path: path.clone(),
5057 },
5058 buffer_id,
5059 );
5060 }
5061
5062 if new_file.entry_id != *entry_id {
5063 self.local_buffer_ids_by_entry_id.remove(entry_id);
5064 self.local_buffer_ids_by_entry_id
5065 .insert(new_file.entry_id, buffer_id);
5066 }
5067
5068 if new_file != *old_file {
5069 if let Some(project_id) = self.remote_id() {
5070 self.client
5071 .send(proto::UpdateBufferFile {
5072 project_id,
5073 buffer_id: buffer_id as u64,
5074 file: Some(new_file.to_proto()),
5075 })
5076 .log_err();
5077 }
5078
5079 buffer.file_updated(Arc::new(new_file), cx).detach();
5080 }
5081 }
5082 });
5083 }
5084
5085 for (buffer, old_file) in renamed_buffers {
5086 self.unregister_buffer_from_language_servers(&buffer, &old_file, cx);
5087 self.detect_language_for_buffer(&buffer, cx);
5088 self.register_buffer_with_language_servers(&buffer, cx);
5089 }
5090 }
5091
5092 fn update_local_worktree_language_servers(
5093 &mut self,
5094 worktree_handle: &ModelHandle<Worktree>,
5095 changes: &[(Arc<Path>, ProjectEntryId, PathChange)],
5096 cx: &mut ModelContext<Self>,
5097 ) {
5098 if changes.is_empty() {
5099 return;
5100 }
5101
5102 let worktree_id = worktree_handle.read(cx).id();
5103 let mut language_server_ids = self
5104 .language_server_ids
5105 .iter()
5106 .filter_map(|((server_worktree_id, _), server_id)| {
5107 (*server_worktree_id == worktree_id).then_some(*server_id)
5108 })
5109 .collect::<Vec<_>>();
5110 language_server_ids.sort();
5111 language_server_ids.dedup();
5112
5113 let abs_path = worktree_handle.read(cx).abs_path();
5114 for server_id in &language_server_ids {
5115 if let Some(server) = self.language_servers.get(server_id) {
5116 if let LanguageServerState::Running {
5117 server,
5118 watched_paths,
5119 ..
5120 } = server
5121 {
5122 if let Some(watched_paths) = watched_paths.get(&worktree_id) {
5123 let params = lsp::DidChangeWatchedFilesParams {
5124 changes: changes
5125 .iter()
5126 .filter_map(|(path, _, change)| {
5127 if !watched_paths.is_match(&path) {
5128 return None;
5129 }
5130 let typ = match change {
5131 PathChange::Loaded => return None,
5132 PathChange::Added => lsp::FileChangeType::CREATED,
5133 PathChange::Removed => lsp::FileChangeType::DELETED,
5134 PathChange::Updated => lsp::FileChangeType::CHANGED,
5135 PathChange::AddedOrUpdated => lsp::FileChangeType::CHANGED,
5136 };
5137 Some(lsp::FileEvent {
5138 uri: lsp::Url::from_file_path(abs_path.join(path)).unwrap(),
5139 typ,
5140 })
5141 })
5142 .collect(),
5143 };
5144
5145 if !params.changes.is_empty() {
5146 server
5147 .notify::<lsp::notification::DidChangeWatchedFiles>(params)
5148 .log_err();
5149 }
5150 }
5151 }
5152 }
5153 }
5154 }
5155
5156 fn update_local_worktree_buffers_git_repos(
5157 &mut self,
5158 worktree_handle: ModelHandle<Worktree>,
5159 changed_repos: &UpdatedGitRepositoriesSet,
5160 cx: &mut ModelContext<Self>,
5161 ) {
5162 debug_assert!(worktree_handle.read(cx).is_local());
5163
5164 // Identify the loading buffers whose containing repository that has changed.
5165 let future_buffers = self
5166 .loading_buffers_by_path
5167 .iter()
5168 .filter_map(|(project_path, receiver)| {
5169 if project_path.worktree_id != worktree_handle.read(cx).id() {
5170 return None;
5171 }
5172 let path = &project_path.path;
5173 changed_repos
5174 .iter()
5175 .find(|(work_dir, _)| path.starts_with(work_dir))?;
5176 let receiver = receiver.clone();
5177 let path = path.clone();
5178 Some(async move {
5179 wait_for_loading_buffer(receiver)
5180 .await
5181 .ok()
5182 .map(|buffer| (buffer, path))
5183 })
5184 })
5185 .collect::<FuturesUnordered<_>>();
5186
5187 // Identify the current buffers whose containing repository has changed.
5188 let current_buffers = self
5189 .opened_buffers
5190 .values()
5191 .filter_map(|buffer| {
5192 let buffer = buffer.upgrade(cx)?;
5193 let file = File::from_dyn(buffer.read(cx).file())?;
5194 if file.worktree != worktree_handle {
5195 return None;
5196 }
5197 let path = file.path();
5198 changed_repos
5199 .iter()
5200 .find(|(work_dir, _)| path.starts_with(work_dir))?;
5201 Some((buffer, path.clone()))
5202 })
5203 .collect::<Vec<_>>();
5204
5205 if future_buffers.len() + current_buffers.len() == 0 {
5206 return;
5207 }
5208
5209 let remote_id = self.remote_id();
5210 let client = self.client.clone();
5211 cx.spawn_weak(move |_, mut cx| async move {
5212 // Wait for all of the buffers to load.
5213 let future_buffers = future_buffers.collect::<Vec<_>>().await;
5214
5215 // Reload the diff base for every buffer whose containing git repository has changed.
5216 let snapshot =
5217 worktree_handle.read_with(&cx, |tree, _| tree.as_local().unwrap().snapshot());
5218 let diff_bases_by_buffer = cx
5219 .background()
5220 .spawn(async move {
5221 future_buffers
5222 .into_iter()
5223 .filter_map(|e| e)
5224 .chain(current_buffers)
5225 .filter_map(|(buffer, path)| {
5226 let (work_directory, repo) =
5227 snapshot.repository_and_work_directory_for_path(&path)?;
5228 let repo = snapshot.get_local_repo(&repo)?;
5229 let relative_path = path.strip_prefix(&work_directory).ok()?;
5230 let base_text = repo.repo_ptr.lock().load_index_text(&relative_path);
5231 Some((buffer, base_text))
5232 })
5233 .collect::<Vec<_>>()
5234 })
5235 .await;
5236
5237 // Assign the new diff bases on all of the buffers.
5238 for (buffer, diff_base) in diff_bases_by_buffer {
5239 let buffer_id = buffer.update(&mut cx, |buffer, cx| {
5240 buffer.set_diff_base(diff_base.clone(), cx);
5241 buffer.remote_id()
5242 });
5243 if let Some(project_id) = remote_id {
5244 client
5245 .send(proto::UpdateDiffBase {
5246 project_id,
5247 buffer_id,
5248 diff_base,
5249 })
5250 .log_err();
5251 }
5252 }
5253 })
5254 .detach();
5255 }
5256
5257 fn update_local_worktree_settings(
5258 &mut self,
5259 worktree: &ModelHandle<Worktree>,
5260 changes: &UpdatedEntriesSet,
5261 cx: &mut ModelContext<Self>,
5262 ) {
5263 let project_id = self.remote_id();
5264 let worktree_id = worktree.id();
5265 let worktree = worktree.read(cx).as_local().unwrap();
5266 let remote_worktree_id = worktree.id();
5267
5268 let mut settings_contents = Vec::new();
5269 for (path, _, change) in changes.iter() {
5270 if path.ends_with(&*LOCAL_SETTINGS_RELATIVE_PATH) {
5271 let settings_dir = Arc::from(
5272 path.ancestors()
5273 .nth(LOCAL_SETTINGS_RELATIVE_PATH.components().count())
5274 .unwrap(),
5275 );
5276 let fs = self.fs.clone();
5277 let removed = *change == PathChange::Removed;
5278 let abs_path = worktree.absolutize(path);
5279 settings_contents.push(async move {
5280 (settings_dir, (!removed).then_some(fs.load(&abs_path).await))
5281 });
5282 }
5283 }
5284
5285 if settings_contents.is_empty() {
5286 return;
5287 }
5288
5289 let client = self.client.clone();
5290 cx.spawn_weak(move |_, mut cx| async move {
5291 let settings_contents: Vec<(Arc<Path>, _)> =
5292 futures::future::join_all(settings_contents).await;
5293 cx.update(|cx| {
5294 cx.update_global::<SettingsStore, _, _>(|store, cx| {
5295 for (directory, file_content) in settings_contents {
5296 let file_content = file_content.and_then(|content| content.log_err());
5297 store
5298 .set_local_settings(
5299 worktree_id,
5300 directory.clone(),
5301 file_content.as_ref().map(String::as_str),
5302 cx,
5303 )
5304 .log_err();
5305 if let Some(remote_id) = project_id {
5306 client
5307 .send(proto::UpdateWorktreeSettings {
5308 project_id: remote_id,
5309 worktree_id: remote_worktree_id.to_proto(),
5310 path: directory.to_string_lossy().into_owned(),
5311 content: file_content,
5312 })
5313 .log_err();
5314 }
5315 }
5316 });
5317 });
5318 })
5319 .detach();
5320 }
5321
5322 pub fn set_active_path(&mut self, entry: Option<ProjectPath>, cx: &mut ModelContext<Self>) {
5323 let new_active_entry = entry.and_then(|project_path| {
5324 let worktree = self.worktree_for_id(project_path.worktree_id, cx)?;
5325 let entry = worktree.read(cx).entry_for_path(project_path.path)?;
5326 Some(entry.id)
5327 });
5328 if new_active_entry != self.active_entry {
5329 self.active_entry = new_active_entry;
5330 cx.emit(Event::ActiveEntryChanged(new_active_entry));
5331 }
5332 }
5333
5334 pub fn language_servers_running_disk_based_diagnostics(
5335 &self,
5336 ) -> impl Iterator<Item = LanguageServerId> + '_ {
5337 self.language_server_statuses
5338 .iter()
5339 .filter_map(|(id, status)| {
5340 if status.has_pending_diagnostic_updates {
5341 Some(*id)
5342 } else {
5343 None
5344 }
5345 })
5346 }
5347
5348 pub fn diagnostic_summary(&self, cx: &AppContext) -> DiagnosticSummary {
5349 let mut summary = DiagnosticSummary::default();
5350 for (_, _, path_summary) in self.diagnostic_summaries(cx) {
5351 summary.error_count += path_summary.error_count;
5352 summary.warning_count += path_summary.warning_count;
5353 }
5354 summary
5355 }
5356
5357 pub fn diagnostic_summaries<'a>(
5358 &'a self,
5359 cx: &'a AppContext,
5360 ) -> impl Iterator<Item = (ProjectPath, LanguageServerId, DiagnosticSummary)> + 'a {
5361 self.visible_worktrees(cx).flat_map(move |worktree| {
5362 let worktree = worktree.read(cx);
5363 let worktree_id = worktree.id();
5364 worktree
5365 .diagnostic_summaries()
5366 .map(move |(path, server_id, summary)| {
5367 (ProjectPath { worktree_id, path }, server_id, summary)
5368 })
5369 })
5370 }
5371
5372 pub fn disk_based_diagnostics_started(
5373 &mut self,
5374 language_server_id: LanguageServerId,
5375 cx: &mut ModelContext<Self>,
5376 ) {
5377 cx.emit(Event::DiskBasedDiagnosticsStarted { language_server_id });
5378 }
5379
5380 pub fn disk_based_diagnostics_finished(
5381 &mut self,
5382 language_server_id: LanguageServerId,
5383 cx: &mut ModelContext<Self>,
5384 ) {
5385 cx.emit(Event::DiskBasedDiagnosticsFinished { language_server_id });
5386 }
5387
5388 pub fn active_entry(&self) -> Option<ProjectEntryId> {
5389 self.active_entry
5390 }
5391
5392 pub fn entry_for_path(&self, path: &ProjectPath, cx: &AppContext) -> Option<Entry> {
5393 self.worktree_for_id(path.worktree_id, cx)?
5394 .read(cx)
5395 .entry_for_path(&path.path)
5396 .cloned()
5397 }
5398
5399 pub fn path_for_entry(&self, entry_id: ProjectEntryId, cx: &AppContext) -> Option<ProjectPath> {
5400 let worktree = self.worktree_for_entry(entry_id, cx)?;
5401 let worktree = worktree.read(cx);
5402 let worktree_id = worktree.id();
5403 let path = worktree.entry_for_id(entry_id)?.path.clone();
5404 Some(ProjectPath { worktree_id, path })
5405 }
5406
5407 pub fn mark_entry_expanded(
5408 &mut self,
5409 worktree_id: WorktreeId,
5410 entry_id: ProjectEntryId,
5411 cx: &mut ModelContext<Self>,
5412 ) -> Option<()> {
5413 if self.is_local() {
5414 let worktree = self.worktree_for_id(worktree_id, cx)?;
5415 worktree.update(cx, |worktree, cx| {
5416 worktree
5417 .as_local_mut()
5418 .unwrap()
5419 .expand_entry_for_id(entry_id, cx);
5420 });
5421 } else if let Some(project_id) = self.remote_id() {
5422 cx.background()
5423 .spawn(self.client.request(proto::ExpandProjectEntry {
5424 project_id,
5425 entry_id: entry_id.to_proto(),
5426 }))
5427 .log_err();
5428 }
5429 Some(())
5430 }
5431
5432 pub fn absolute_path(&self, project_path: &ProjectPath, cx: &AppContext) -> Option<PathBuf> {
5433 let workspace_root = self
5434 .worktree_for_id(project_path.worktree_id, cx)?
5435 .read(cx)
5436 .abs_path();
5437 let project_path = project_path.path.as_ref();
5438
5439 Some(if project_path == Path::new("") {
5440 workspace_root.to_path_buf()
5441 } else {
5442 workspace_root.join(project_path)
5443 })
5444 }
5445
5446 // RPC message handlers
5447
5448 async fn handle_unshare_project(
5449 this: ModelHandle<Self>,
5450 _: TypedEnvelope<proto::UnshareProject>,
5451 _: Arc<Client>,
5452 mut cx: AsyncAppContext,
5453 ) -> Result<()> {
5454 this.update(&mut cx, |this, cx| {
5455 if this.is_local() {
5456 this.unshare(cx)?;
5457 } else {
5458 this.disconnected_from_host(cx);
5459 }
5460 Ok(())
5461 })
5462 }
5463
5464 async fn handle_add_collaborator(
5465 this: ModelHandle<Self>,
5466 mut envelope: TypedEnvelope<proto::AddProjectCollaborator>,
5467 _: Arc<Client>,
5468 mut cx: AsyncAppContext,
5469 ) -> Result<()> {
5470 let collaborator = envelope
5471 .payload
5472 .collaborator
5473 .take()
5474 .ok_or_else(|| anyhow!("empty collaborator"))?;
5475
5476 let collaborator = Collaborator::from_proto(collaborator)?;
5477 this.update(&mut cx, |this, cx| {
5478 this.shared_buffers.remove(&collaborator.peer_id);
5479 this.collaborators
5480 .insert(collaborator.peer_id, collaborator);
5481 cx.notify();
5482 });
5483
5484 Ok(())
5485 }
5486
5487 async fn handle_update_project_collaborator(
5488 this: ModelHandle<Self>,
5489 envelope: TypedEnvelope<proto::UpdateProjectCollaborator>,
5490 _: Arc<Client>,
5491 mut cx: AsyncAppContext,
5492 ) -> Result<()> {
5493 let old_peer_id = envelope
5494 .payload
5495 .old_peer_id
5496 .ok_or_else(|| anyhow!("missing old peer id"))?;
5497 let new_peer_id = envelope
5498 .payload
5499 .new_peer_id
5500 .ok_or_else(|| anyhow!("missing new peer id"))?;
5501 this.update(&mut cx, |this, cx| {
5502 let collaborator = this
5503 .collaborators
5504 .remove(&old_peer_id)
5505 .ok_or_else(|| anyhow!("received UpdateProjectCollaborator for unknown peer"))?;
5506 let is_host = collaborator.replica_id == 0;
5507 this.collaborators.insert(new_peer_id, collaborator);
5508
5509 let buffers = this.shared_buffers.remove(&old_peer_id);
5510 log::info!(
5511 "peer {} became {}. moving buffers {:?}",
5512 old_peer_id,
5513 new_peer_id,
5514 &buffers
5515 );
5516 if let Some(buffers) = buffers {
5517 this.shared_buffers.insert(new_peer_id, buffers);
5518 }
5519
5520 if is_host {
5521 this.opened_buffers
5522 .retain(|_, buffer| !matches!(buffer, OpenBuffer::Operations(_)));
5523 this.buffer_ordered_messages_tx
5524 .unbounded_send(BufferOrderedMessage::Resync)
5525 .unwrap();
5526 }
5527
5528 cx.emit(Event::CollaboratorUpdated {
5529 old_peer_id,
5530 new_peer_id,
5531 });
5532 cx.notify();
5533 Ok(())
5534 })
5535 }
5536
5537 async fn handle_remove_collaborator(
5538 this: ModelHandle<Self>,
5539 envelope: TypedEnvelope<proto::RemoveProjectCollaborator>,
5540 _: Arc<Client>,
5541 mut cx: AsyncAppContext,
5542 ) -> Result<()> {
5543 this.update(&mut cx, |this, cx| {
5544 let peer_id = envelope
5545 .payload
5546 .peer_id
5547 .ok_or_else(|| anyhow!("invalid peer id"))?;
5548 let replica_id = this
5549 .collaborators
5550 .remove(&peer_id)
5551 .ok_or_else(|| anyhow!("unknown peer {:?}", peer_id))?
5552 .replica_id;
5553 for buffer in this.opened_buffers.values() {
5554 if let Some(buffer) = buffer.upgrade(cx) {
5555 buffer.update(cx, |buffer, cx| buffer.remove_peer(replica_id, cx));
5556 }
5557 }
5558 this.shared_buffers.remove(&peer_id);
5559
5560 cx.emit(Event::CollaboratorLeft(peer_id));
5561 cx.notify();
5562 Ok(())
5563 })
5564 }
5565
5566 async fn handle_update_project(
5567 this: ModelHandle<Self>,
5568 envelope: TypedEnvelope<proto::UpdateProject>,
5569 _: Arc<Client>,
5570 mut cx: AsyncAppContext,
5571 ) -> Result<()> {
5572 this.update(&mut cx, |this, cx| {
5573 // Don't handle messages that were sent before the response to us joining the project
5574 if envelope.message_id > this.join_project_response_message_id {
5575 this.set_worktrees_from_proto(envelope.payload.worktrees, cx)?;
5576 }
5577 Ok(())
5578 })
5579 }
5580
5581 async fn handle_update_worktree(
5582 this: ModelHandle<Self>,
5583 envelope: TypedEnvelope<proto::UpdateWorktree>,
5584 _: Arc<Client>,
5585 mut cx: AsyncAppContext,
5586 ) -> Result<()> {
5587 this.update(&mut cx, |this, cx| {
5588 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
5589 if let Some(worktree) = this.worktree_for_id(worktree_id, cx) {
5590 worktree.update(cx, |worktree, _| {
5591 let worktree = worktree.as_remote_mut().unwrap();
5592 worktree.update_from_remote(envelope.payload);
5593 });
5594 }
5595 Ok(())
5596 })
5597 }
5598
5599 async fn handle_update_worktree_settings(
5600 this: ModelHandle<Self>,
5601 envelope: TypedEnvelope<proto::UpdateWorktreeSettings>,
5602 _: Arc<Client>,
5603 mut cx: AsyncAppContext,
5604 ) -> Result<()> {
5605 this.update(&mut cx, |this, cx| {
5606 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
5607 if let Some(worktree) = this.worktree_for_id(worktree_id, cx) {
5608 cx.update_global::<SettingsStore, _, _>(|store, cx| {
5609 store
5610 .set_local_settings(
5611 worktree.id(),
5612 PathBuf::from(&envelope.payload.path).into(),
5613 envelope.payload.content.as_ref().map(String::as_str),
5614 cx,
5615 )
5616 .log_err();
5617 });
5618 }
5619 Ok(())
5620 })
5621 }
5622
5623 async fn handle_create_project_entry(
5624 this: ModelHandle<Self>,
5625 envelope: TypedEnvelope<proto::CreateProjectEntry>,
5626 _: Arc<Client>,
5627 mut cx: AsyncAppContext,
5628 ) -> Result<proto::ProjectEntryResponse> {
5629 let worktree = this.update(&mut cx, |this, cx| {
5630 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
5631 this.worktree_for_id(worktree_id, cx)
5632 .ok_or_else(|| anyhow!("worktree not found"))
5633 })?;
5634 let worktree_scan_id = worktree.read_with(&cx, |worktree, _| worktree.scan_id());
5635 let entry = worktree
5636 .update(&mut cx, |worktree, cx| {
5637 let worktree = worktree.as_local_mut().unwrap();
5638 let path = PathBuf::from(envelope.payload.path);
5639 worktree.create_entry(path, envelope.payload.is_directory, cx)
5640 })
5641 .await?;
5642 Ok(proto::ProjectEntryResponse {
5643 entry: Some((&entry).into()),
5644 worktree_scan_id: worktree_scan_id as u64,
5645 })
5646 }
5647
5648 async fn handle_rename_project_entry(
5649 this: ModelHandle<Self>,
5650 envelope: TypedEnvelope<proto::RenameProjectEntry>,
5651 _: Arc<Client>,
5652 mut cx: AsyncAppContext,
5653 ) -> Result<proto::ProjectEntryResponse> {
5654 let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
5655 let worktree = this.read_with(&cx, |this, cx| {
5656 this.worktree_for_entry(entry_id, cx)
5657 .ok_or_else(|| anyhow!("worktree not found"))
5658 })?;
5659 let worktree_scan_id = worktree.read_with(&cx, |worktree, _| worktree.scan_id());
5660 let entry = worktree
5661 .update(&mut cx, |worktree, cx| {
5662 let new_path = PathBuf::from(envelope.payload.new_path);
5663 worktree
5664 .as_local_mut()
5665 .unwrap()
5666 .rename_entry(entry_id, new_path, cx)
5667 .ok_or_else(|| anyhow!("invalid entry"))
5668 })?
5669 .await?;
5670 Ok(proto::ProjectEntryResponse {
5671 entry: Some((&entry).into()),
5672 worktree_scan_id: worktree_scan_id as u64,
5673 })
5674 }
5675
5676 async fn handle_copy_project_entry(
5677 this: ModelHandle<Self>,
5678 envelope: TypedEnvelope<proto::CopyProjectEntry>,
5679 _: Arc<Client>,
5680 mut cx: AsyncAppContext,
5681 ) -> Result<proto::ProjectEntryResponse> {
5682 let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
5683 let worktree = this.read_with(&cx, |this, cx| {
5684 this.worktree_for_entry(entry_id, cx)
5685 .ok_or_else(|| anyhow!("worktree not found"))
5686 })?;
5687 let worktree_scan_id = worktree.read_with(&cx, |worktree, _| worktree.scan_id());
5688 let entry = worktree
5689 .update(&mut cx, |worktree, cx| {
5690 let new_path = PathBuf::from(envelope.payload.new_path);
5691 worktree
5692 .as_local_mut()
5693 .unwrap()
5694 .copy_entry(entry_id, new_path, cx)
5695 .ok_or_else(|| anyhow!("invalid entry"))
5696 })?
5697 .await?;
5698 Ok(proto::ProjectEntryResponse {
5699 entry: Some((&entry).into()),
5700 worktree_scan_id: worktree_scan_id as u64,
5701 })
5702 }
5703
5704 async fn handle_delete_project_entry(
5705 this: ModelHandle<Self>,
5706 envelope: TypedEnvelope<proto::DeleteProjectEntry>,
5707 _: Arc<Client>,
5708 mut cx: AsyncAppContext,
5709 ) -> Result<proto::ProjectEntryResponse> {
5710 let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
5711
5712 this.update(&mut cx, |_, cx| cx.emit(Event::DeletedEntry(entry_id)));
5713
5714 let worktree = this.read_with(&cx, |this, cx| {
5715 this.worktree_for_entry(entry_id, cx)
5716 .ok_or_else(|| anyhow!("worktree not found"))
5717 })?;
5718 let worktree_scan_id = worktree.read_with(&cx, |worktree, _| worktree.scan_id());
5719 worktree
5720 .update(&mut cx, |worktree, cx| {
5721 worktree
5722 .as_local_mut()
5723 .unwrap()
5724 .delete_entry(entry_id, cx)
5725 .ok_or_else(|| anyhow!("invalid entry"))
5726 })?
5727 .await?;
5728 Ok(proto::ProjectEntryResponse {
5729 entry: None,
5730 worktree_scan_id: worktree_scan_id as u64,
5731 })
5732 }
5733
5734 async fn handle_expand_project_entry(
5735 this: ModelHandle<Self>,
5736 envelope: TypedEnvelope<proto::ExpandProjectEntry>,
5737 _: Arc<Client>,
5738 mut cx: AsyncAppContext,
5739 ) -> Result<proto::Ack> {
5740 let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
5741 let worktree = this
5742 .read_with(&cx, |this, cx| this.worktree_for_entry(entry_id, cx))
5743 .ok_or_else(|| anyhow!("invalid request"))?;
5744 worktree.update(&mut cx, |worktree, cx| {
5745 worktree
5746 .as_local_mut()
5747 .unwrap()
5748 .expand_entry_for_id(entry_id, cx)
5749 });
5750 Ok(proto::Ack {})
5751 }
5752
5753 async fn handle_update_diagnostic_summary(
5754 this: ModelHandle<Self>,
5755 envelope: TypedEnvelope<proto::UpdateDiagnosticSummary>,
5756 _: Arc<Client>,
5757 mut cx: AsyncAppContext,
5758 ) -> Result<()> {
5759 this.update(&mut cx, |this, cx| {
5760 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
5761 if let Some(worktree) = this.worktree_for_id(worktree_id, cx) {
5762 if let Some(summary) = envelope.payload.summary {
5763 let project_path = ProjectPath {
5764 worktree_id,
5765 path: Path::new(&summary.path).into(),
5766 };
5767 worktree.update(cx, |worktree, _| {
5768 worktree
5769 .as_remote_mut()
5770 .unwrap()
5771 .update_diagnostic_summary(project_path.path.clone(), &summary);
5772 });
5773 cx.emit(Event::DiagnosticsUpdated {
5774 language_server_id: LanguageServerId(summary.language_server_id as usize),
5775 path: project_path,
5776 });
5777 }
5778 }
5779 Ok(())
5780 })
5781 }
5782
5783 async fn handle_start_language_server(
5784 this: ModelHandle<Self>,
5785 envelope: TypedEnvelope<proto::StartLanguageServer>,
5786 _: Arc<Client>,
5787 mut cx: AsyncAppContext,
5788 ) -> Result<()> {
5789 let server = envelope
5790 .payload
5791 .server
5792 .ok_or_else(|| anyhow!("invalid server"))?;
5793 this.update(&mut cx, |this, cx| {
5794 this.language_server_statuses.insert(
5795 LanguageServerId(server.id as usize),
5796 LanguageServerStatus {
5797 name: server.name,
5798 pending_work: Default::default(),
5799 has_pending_diagnostic_updates: false,
5800 progress_tokens: Default::default(),
5801 },
5802 );
5803 cx.notify();
5804 });
5805 Ok(())
5806 }
5807
5808 async fn handle_update_language_server(
5809 this: ModelHandle<Self>,
5810 envelope: TypedEnvelope<proto::UpdateLanguageServer>,
5811 _: Arc<Client>,
5812 mut cx: AsyncAppContext,
5813 ) -> Result<()> {
5814 this.update(&mut cx, |this, cx| {
5815 let language_server_id = LanguageServerId(envelope.payload.language_server_id as usize);
5816
5817 match envelope
5818 .payload
5819 .variant
5820 .ok_or_else(|| anyhow!("invalid variant"))?
5821 {
5822 proto::update_language_server::Variant::WorkStart(payload) => {
5823 this.on_lsp_work_start(
5824 language_server_id,
5825 payload.token,
5826 LanguageServerProgress {
5827 message: payload.message,
5828 percentage: payload.percentage.map(|p| p as usize),
5829 last_update_at: Instant::now(),
5830 },
5831 cx,
5832 );
5833 }
5834
5835 proto::update_language_server::Variant::WorkProgress(payload) => {
5836 this.on_lsp_work_progress(
5837 language_server_id,
5838 payload.token,
5839 LanguageServerProgress {
5840 message: payload.message,
5841 percentage: payload.percentage.map(|p| p as usize),
5842 last_update_at: Instant::now(),
5843 },
5844 cx,
5845 );
5846 }
5847
5848 proto::update_language_server::Variant::WorkEnd(payload) => {
5849 this.on_lsp_work_end(language_server_id, payload.token, cx);
5850 }
5851
5852 proto::update_language_server::Variant::DiskBasedDiagnosticsUpdating(_) => {
5853 this.disk_based_diagnostics_started(language_server_id, cx);
5854 }
5855
5856 proto::update_language_server::Variant::DiskBasedDiagnosticsUpdated(_) => {
5857 this.disk_based_diagnostics_finished(language_server_id, cx)
5858 }
5859 }
5860
5861 Ok(())
5862 })
5863 }
5864
5865 async fn handle_update_buffer(
5866 this: ModelHandle<Self>,
5867 envelope: TypedEnvelope<proto::UpdateBuffer>,
5868 _: Arc<Client>,
5869 mut cx: AsyncAppContext,
5870 ) -> Result<proto::Ack> {
5871 this.update(&mut cx, |this, cx| {
5872 let payload = envelope.payload.clone();
5873 let buffer_id = payload.buffer_id;
5874 let ops = payload
5875 .operations
5876 .into_iter()
5877 .map(language::proto::deserialize_operation)
5878 .collect::<Result<Vec<_>, _>>()?;
5879 let is_remote = this.is_remote();
5880 match this.opened_buffers.entry(buffer_id) {
5881 hash_map::Entry::Occupied(mut e) => match e.get_mut() {
5882 OpenBuffer::Strong(buffer) => {
5883 buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx))?;
5884 }
5885 OpenBuffer::Operations(operations) => operations.extend_from_slice(&ops),
5886 OpenBuffer::Weak(_) => {}
5887 },
5888 hash_map::Entry::Vacant(e) => {
5889 assert!(
5890 is_remote,
5891 "received buffer update from {:?}",
5892 envelope.original_sender_id
5893 );
5894 e.insert(OpenBuffer::Operations(ops));
5895 }
5896 }
5897 Ok(proto::Ack {})
5898 })
5899 }
5900
5901 async fn handle_create_buffer_for_peer(
5902 this: ModelHandle<Self>,
5903 envelope: TypedEnvelope<proto::CreateBufferForPeer>,
5904 _: Arc<Client>,
5905 mut cx: AsyncAppContext,
5906 ) -> Result<()> {
5907 this.update(&mut cx, |this, cx| {
5908 match envelope
5909 .payload
5910 .variant
5911 .ok_or_else(|| anyhow!("missing variant"))?
5912 {
5913 proto::create_buffer_for_peer::Variant::State(mut state) => {
5914 let mut buffer_file = None;
5915 if let Some(file) = state.file.take() {
5916 let worktree_id = WorktreeId::from_proto(file.worktree_id);
5917 let worktree = this.worktree_for_id(worktree_id, cx).ok_or_else(|| {
5918 anyhow!("no worktree found for id {}", file.worktree_id)
5919 })?;
5920 buffer_file = Some(Arc::new(File::from_proto(file, worktree.clone(), cx)?)
5921 as Arc<dyn language::File>);
5922 }
5923
5924 let buffer_id = state.id;
5925 let buffer = cx.add_model(|_| {
5926 Buffer::from_proto(this.replica_id(), state, buffer_file).unwrap()
5927 });
5928 this.incomplete_remote_buffers
5929 .insert(buffer_id, Some(buffer));
5930 }
5931 proto::create_buffer_for_peer::Variant::Chunk(chunk) => {
5932 let buffer = this
5933 .incomplete_remote_buffers
5934 .get(&chunk.buffer_id)
5935 .cloned()
5936 .flatten()
5937 .ok_or_else(|| {
5938 anyhow!(
5939 "received chunk for buffer {} without initial state",
5940 chunk.buffer_id
5941 )
5942 })?;
5943 let operations = chunk
5944 .operations
5945 .into_iter()
5946 .map(language::proto::deserialize_operation)
5947 .collect::<Result<Vec<_>>>()?;
5948 buffer.update(cx, |buffer, cx| buffer.apply_ops(operations, cx))?;
5949
5950 if chunk.is_last {
5951 this.incomplete_remote_buffers.remove(&chunk.buffer_id);
5952 this.register_buffer(&buffer, cx)?;
5953 }
5954 }
5955 }
5956
5957 Ok(())
5958 })
5959 }
5960
5961 async fn handle_update_diff_base(
5962 this: ModelHandle<Self>,
5963 envelope: TypedEnvelope<proto::UpdateDiffBase>,
5964 _: Arc<Client>,
5965 mut cx: AsyncAppContext,
5966 ) -> Result<()> {
5967 this.update(&mut cx, |this, cx| {
5968 let buffer_id = envelope.payload.buffer_id;
5969 let diff_base = envelope.payload.diff_base;
5970 if let Some(buffer) = this
5971 .opened_buffers
5972 .get_mut(&buffer_id)
5973 .and_then(|b| b.upgrade(cx))
5974 .or_else(|| {
5975 this.incomplete_remote_buffers
5976 .get(&buffer_id)
5977 .cloned()
5978 .flatten()
5979 })
5980 {
5981 buffer.update(cx, |buffer, cx| buffer.set_diff_base(diff_base, cx));
5982 }
5983 Ok(())
5984 })
5985 }
5986
5987 async fn handle_update_buffer_file(
5988 this: ModelHandle<Self>,
5989 envelope: TypedEnvelope<proto::UpdateBufferFile>,
5990 _: Arc<Client>,
5991 mut cx: AsyncAppContext,
5992 ) -> Result<()> {
5993 let buffer_id = envelope.payload.buffer_id;
5994
5995 this.update(&mut cx, |this, cx| {
5996 let payload = envelope.payload.clone();
5997 if let Some(buffer) = this
5998 .opened_buffers
5999 .get(&buffer_id)
6000 .and_then(|b| b.upgrade(cx))
6001 .or_else(|| {
6002 this.incomplete_remote_buffers
6003 .get(&buffer_id)
6004 .cloned()
6005 .flatten()
6006 })
6007 {
6008 let file = payload.file.ok_or_else(|| anyhow!("invalid file"))?;
6009 let worktree = this
6010 .worktree_for_id(WorktreeId::from_proto(file.worktree_id), cx)
6011 .ok_or_else(|| anyhow!("no such worktree"))?;
6012 let file = File::from_proto(file, worktree, cx)?;
6013 buffer.update(cx, |buffer, cx| {
6014 buffer.file_updated(Arc::new(file), cx).detach();
6015 });
6016 this.detect_language_for_buffer(&buffer, cx);
6017 }
6018 Ok(())
6019 })
6020 }
6021
6022 async fn handle_save_buffer(
6023 this: ModelHandle<Self>,
6024 envelope: TypedEnvelope<proto::SaveBuffer>,
6025 _: Arc<Client>,
6026 mut cx: AsyncAppContext,
6027 ) -> Result<proto::BufferSaved> {
6028 let buffer_id = envelope.payload.buffer_id;
6029 let (project_id, buffer) = this.update(&mut cx, |this, cx| {
6030 let project_id = this.remote_id().ok_or_else(|| anyhow!("not connected"))?;
6031 let buffer = this
6032 .opened_buffers
6033 .get(&buffer_id)
6034 .and_then(|buffer| buffer.upgrade(cx))
6035 .ok_or_else(|| anyhow!("unknown buffer id {}", buffer_id))?;
6036 anyhow::Ok((project_id, buffer))
6037 })?;
6038 buffer
6039 .update(&mut cx, |buffer, _| {
6040 buffer.wait_for_version(deserialize_version(&envelope.payload.version))
6041 })
6042 .await?;
6043 let buffer_id = buffer.read_with(&cx, |buffer, _| buffer.remote_id());
6044
6045 this.update(&mut cx, |this, cx| this.save_buffer(buffer.clone(), cx))
6046 .await?;
6047 Ok(buffer.read_with(&cx, |buffer, _| proto::BufferSaved {
6048 project_id,
6049 buffer_id,
6050 version: serialize_version(buffer.saved_version()),
6051 mtime: Some(buffer.saved_mtime().into()),
6052 fingerprint: language::proto::serialize_fingerprint(buffer.saved_version_fingerprint()),
6053 }))
6054 }
6055
6056 async fn handle_reload_buffers(
6057 this: ModelHandle<Self>,
6058 envelope: TypedEnvelope<proto::ReloadBuffers>,
6059 _: Arc<Client>,
6060 mut cx: AsyncAppContext,
6061 ) -> Result<proto::ReloadBuffersResponse> {
6062 let sender_id = envelope.original_sender_id()?;
6063 let reload = this.update(&mut cx, |this, cx| {
6064 let mut buffers = HashSet::default();
6065 for buffer_id in &envelope.payload.buffer_ids {
6066 buffers.insert(
6067 this.opened_buffers
6068 .get(buffer_id)
6069 .and_then(|buffer| buffer.upgrade(cx))
6070 .ok_or_else(|| anyhow!("unknown buffer id {}", buffer_id))?,
6071 );
6072 }
6073 Ok::<_, anyhow::Error>(this.reload_buffers(buffers, false, cx))
6074 })?;
6075
6076 let project_transaction = reload.await?;
6077 let project_transaction = this.update(&mut cx, |this, cx| {
6078 this.serialize_project_transaction_for_peer(project_transaction, sender_id, cx)
6079 });
6080 Ok(proto::ReloadBuffersResponse {
6081 transaction: Some(project_transaction),
6082 })
6083 }
6084
6085 async fn handle_synchronize_buffers(
6086 this: ModelHandle<Self>,
6087 envelope: TypedEnvelope<proto::SynchronizeBuffers>,
6088 _: Arc<Client>,
6089 mut cx: AsyncAppContext,
6090 ) -> Result<proto::SynchronizeBuffersResponse> {
6091 let project_id = envelope.payload.project_id;
6092 let mut response = proto::SynchronizeBuffersResponse {
6093 buffers: Default::default(),
6094 };
6095
6096 this.update(&mut cx, |this, cx| {
6097 let Some(guest_id) = envelope.original_sender_id else {
6098 error!("missing original_sender_id on SynchronizeBuffers request");
6099 return;
6100 };
6101
6102 this.shared_buffers.entry(guest_id).or_default().clear();
6103 for buffer in envelope.payload.buffers {
6104 let buffer_id = buffer.id;
6105 let remote_version = language::proto::deserialize_version(&buffer.version);
6106 if let Some(buffer) = this.buffer_for_id(buffer_id, cx) {
6107 this.shared_buffers
6108 .entry(guest_id)
6109 .or_default()
6110 .insert(buffer_id);
6111
6112 let buffer = buffer.read(cx);
6113 response.buffers.push(proto::BufferVersion {
6114 id: buffer_id,
6115 version: language::proto::serialize_version(&buffer.version),
6116 });
6117
6118 let operations = buffer.serialize_ops(Some(remote_version), cx);
6119 let client = this.client.clone();
6120 if let Some(file) = buffer.file() {
6121 client
6122 .send(proto::UpdateBufferFile {
6123 project_id,
6124 buffer_id: buffer_id as u64,
6125 file: Some(file.to_proto()),
6126 })
6127 .log_err();
6128 }
6129
6130 client
6131 .send(proto::UpdateDiffBase {
6132 project_id,
6133 buffer_id: buffer_id as u64,
6134 diff_base: buffer.diff_base().map(Into::into),
6135 })
6136 .log_err();
6137
6138 client
6139 .send(proto::BufferReloaded {
6140 project_id,
6141 buffer_id,
6142 version: language::proto::serialize_version(buffer.saved_version()),
6143 mtime: Some(buffer.saved_mtime().into()),
6144 fingerprint: language::proto::serialize_fingerprint(
6145 buffer.saved_version_fingerprint(),
6146 ),
6147 line_ending: language::proto::serialize_line_ending(
6148 buffer.line_ending(),
6149 ) as i32,
6150 })
6151 .log_err();
6152
6153 cx.background()
6154 .spawn(
6155 async move {
6156 let operations = operations.await;
6157 for chunk in split_operations(operations) {
6158 client
6159 .request(proto::UpdateBuffer {
6160 project_id,
6161 buffer_id,
6162 operations: chunk,
6163 })
6164 .await?;
6165 }
6166 anyhow::Ok(())
6167 }
6168 .log_err(),
6169 )
6170 .detach();
6171 }
6172 }
6173 });
6174
6175 Ok(response)
6176 }
6177
6178 async fn handle_format_buffers(
6179 this: ModelHandle<Self>,
6180 envelope: TypedEnvelope<proto::FormatBuffers>,
6181 _: Arc<Client>,
6182 mut cx: AsyncAppContext,
6183 ) -> Result<proto::FormatBuffersResponse> {
6184 let sender_id = envelope.original_sender_id()?;
6185 let format = this.update(&mut cx, |this, cx| {
6186 let mut buffers = HashSet::default();
6187 for buffer_id in &envelope.payload.buffer_ids {
6188 buffers.insert(
6189 this.opened_buffers
6190 .get(buffer_id)
6191 .and_then(|buffer| buffer.upgrade(cx))
6192 .ok_or_else(|| anyhow!("unknown buffer id {}", buffer_id))?,
6193 );
6194 }
6195 let trigger = FormatTrigger::from_proto(envelope.payload.trigger);
6196 Ok::<_, anyhow::Error>(this.format(buffers, false, trigger, cx))
6197 })?;
6198
6199 let project_transaction = format.await?;
6200 let project_transaction = this.update(&mut cx, |this, cx| {
6201 this.serialize_project_transaction_for_peer(project_transaction, sender_id, cx)
6202 });
6203 Ok(proto::FormatBuffersResponse {
6204 transaction: Some(project_transaction),
6205 })
6206 }
6207
6208 async fn handle_apply_additional_edits_for_completion(
6209 this: ModelHandle<Self>,
6210 envelope: TypedEnvelope<proto::ApplyCompletionAdditionalEdits>,
6211 _: Arc<Client>,
6212 mut cx: AsyncAppContext,
6213 ) -> Result<proto::ApplyCompletionAdditionalEditsResponse> {
6214 let (buffer, completion) = this.update(&mut cx, |this, cx| {
6215 let buffer = this
6216 .opened_buffers
6217 .get(&envelope.payload.buffer_id)
6218 .and_then(|buffer| buffer.upgrade(cx))
6219 .ok_or_else(|| anyhow!("unknown buffer id {}", envelope.payload.buffer_id))?;
6220 let language = buffer.read(cx).language();
6221 let completion = language::proto::deserialize_completion(
6222 envelope
6223 .payload
6224 .completion
6225 .ok_or_else(|| anyhow!("invalid completion"))?,
6226 language.cloned(),
6227 );
6228 Ok::<_, anyhow::Error>((buffer, completion))
6229 })?;
6230
6231 let completion = completion.await?;
6232
6233 let apply_additional_edits = this.update(&mut cx, |this, cx| {
6234 this.apply_additional_edits_for_completion(buffer, completion, false, cx)
6235 });
6236
6237 Ok(proto::ApplyCompletionAdditionalEditsResponse {
6238 transaction: apply_additional_edits
6239 .await?
6240 .as_ref()
6241 .map(language::proto::serialize_transaction),
6242 })
6243 }
6244
6245 async fn handle_apply_code_action(
6246 this: ModelHandle<Self>,
6247 envelope: TypedEnvelope<proto::ApplyCodeAction>,
6248 _: Arc<Client>,
6249 mut cx: AsyncAppContext,
6250 ) -> Result<proto::ApplyCodeActionResponse> {
6251 let sender_id = envelope.original_sender_id()?;
6252 let action = language::proto::deserialize_code_action(
6253 envelope
6254 .payload
6255 .action
6256 .ok_or_else(|| anyhow!("invalid action"))?,
6257 )?;
6258 let apply_code_action = this.update(&mut cx, |this, cx| {
6259 let buffer = this
6260 .opened_buffers
6261 .get(&envelope.payload.buffer_id)
6262 .and_then(|buffer| buffer.upgrade(cx))
6263 .ok_or_else(|| anyhow!("unknown buffer id {}", envelope.payload.buffer_id))?;
6264 Ok::<_, anyhow::Error>(this.apply_code_action(buffer, action, false, cx))
6265 })?;
6266
6267 let project_transaction = apply_code_action.await?;
6268 let project_transaction = this.update(&mut cx, |this, cx| {
6269 this.serialize_project_transaction_for_peer(project_transaction, sender_id, cx)
6270 });
6271 Ok(proto::ApplyCodeActionResponse {
6272 transaction: Some(project_transaction),
6273 })
6274 }
6275
6276 async fn handle_on_type_formatting(
6277 this: ModelHandle<Self>,
6278 envelope: TypedEnvelope<proto::OnTypeFormatting>,
6279 _: Arc<Client>,
6280 mut cx: AsyncAppContext,
6281 ) -> Result<proto::OnTypeFormattingResponse> {
6282 let on_type_formatting = this.update(&mut cx, |this, cx| {
6283 let buffer = this
6284 .opened_buffers
6285 .get(&envelope.payload.buffer_id)
6286 .and_then(|buffer| buffer.upgrade(cx))
6287 .ok_or_else(|| anyhow!("unknown buffer id {}", envelope.payload.buffer_id))?;
6288 let position = envelope
6289 .payload
6290 .position
6291 .and_then(deserialize_anchor)
6292 .ok_or_else(|| anyhow!("invalid position"))?;
6293 Ok::<_, anyhow::Error>(this.apply_on_type_formatting(
6294 buffer,
6295 position,
6296 envelope.payload.trigger.clone(),
6297 cx,
6298 ))
6299 })?;
6300
6301 let transaction = on_type_formatting
6302 .await?
6303 .as_ref()
6304 .map(language::proto::serialize_transaction);
6305 Ok(proto::OnTypeFormattingResponse { transaction })
6306 }
6307
6308 async fn handle_lsp_command<T: LspCommand>(
6309 this: ModelHandle<Self>,
6310 envelope: TypedEnvelope<T::ProtoRequest>,
6311 _: Arc<Client>,
6312 mut cx: AsyncAppContext,
6313 ) -> Result<<T::ProtoRequest as proto::RequestMessage>::Response>
6314 where
6315 <T::LspRequest as lsp::request::Request>::Result: Send,
6316 {
6317 let sender_id = envelope.original_sender_id()?;
6318 let buffer_id = T::buffer_id_from_proto(&envelope.payload);
6319 let buffer_handle = this.read_with(&cx, |this, _| {
6320 this.opened_buffers
6321 .get(&buffer_id)
6322 .and_then(|buffer| buffer.upgrade(&cx))
6323 .ok_or_else(|| anyhow!("unknown buffer id {}", buffer_id))
6324 })?;
6325 let request = T::from_proto(
6326 envelope.payload,
6327 this.clone(),
6328 buffer_handle.clone(),
6329 cx.clone(),
6330 )
6331 .await?;
6332 let buffer_version = buffer_handle.read_with(&cx, |buffer, _| buffer.version());
6333 let response = this
6334 .update(&mut cx, |this, cx| {
6335 this.request_lsp(buffer_handle, request, cx)
6336 })
6337 .await?;
6338 this.update(&mut cx, |this, cx| {
6339 Ok(T::response_to_proto(
6340 response,
6341 this,
6342 sender_id,
6343 &buffer_version,
6344 cx,
6345 ))
6346 })
6347 }
6348
6349 async fn handle_get_project_symbols(
6350 this: ModelHandle<Self>,
6351 envelope: TypedEnvelope<proto::GetProjectSymbols>,
6352 _: Arc<Client>,
6353 mut cx: AsyncAppContext,
6354 ) -> Result<proto::GetProjectSymbolsResponse> {
6355 let symbols = this
6356 .update(&mut cx, |this, cx| {
6357 this.symbols(&envelope.payload.query, cx)
6358 })
6359 .await?;
6360
6361 Ok(proto::GetProjectSymbolsResponse {
6362 symbols: symbols.iter().map(serialize_symbol).collect(),
6363 })
6364 }
6365
6366 async fn handle_search_project(
6367 this: ModelHandle<Self>,
6368 envelope: TypedEnvelope<proto::SearchProject>,
6369 _: Arc<Client>,
6370 mut cx: AsyncAppContext,
6371 ) -> Result<proto::SearchProjectResponse> {
6372 let peer_id = envelope.original_sender_id()?;
6373 let query = SearchQuery::from_proto(envelope.payload)?;
6374 let result = this
6375 .update(&mut cx, |this, cx| this.search(query, cx))
6376 .await?;
6377
6378 this.update(&mut cx, |this, cx| {
6379 let mut locations = Vec::new();
6380 for (buffer, ranges) in result {
6381 for range in ranges {
6382 let start = serialize_anchor(&range.start);
6383 let end = serialize_anchor(&range.end);
6384 let buffer_id = this.create_buffer_for_peer(&buffer, peer_id, cx);
6385 locations.push(proto::Location {
6386 buffer_id,
6387 start: Some(start),
6388 end: Some(end),
6389 });
6390 }
6391 }
6392 Ok(proto::SearchProjectResponse { locations })
6393 })
6394 }
6395
6396 async fn handle_open_buffer_for_symbol(
6397 this: ModelHandle<Self>,
6398 envelope: TypedEnvelope<proto::OpenBufferForSymbol>,
6399 _: Arc<Client>,
6400 mut cx: AsyncAppContext,
6401 ) -> Result<proto::OpenBufferForSymbolResponse> {
6402 let peer_id = envelope.original_sender_id()?;
6403 let symbol = envelope
6404 .payload
6405 .symbol
6406 .ok_or_else(|| anyhow!("invalid symbol"))?;
6407 let symbol = this
6408 .read_with(&cx, |this, _| this.deserialize_symbol(symbol))
6409 .await?;
6410 let symbol = this.read_with(&cx, |this, _| {
6411 let signature = this.symbol_signature(&symbol.path);
6412 if signature == symbol.signature {
6413 Ok(symbol)
6414 } else {
6415 Err(anyhow!("invalid symbol signature"))
6416 }
6417 })?;
6418 let buffer = this
6419 .update(&mut cx, |this, cx| this.open_buffer_for_symbol(&symbol, cx))
6420 .await?;
6421
6422 Ok(proto::OpenBufferForSymbolResponse {
6423 buffer_id: this.update(&mut cx, |this, cx| {
6424 this.create_buffer_for_peer(&buffer, peer_id, cx)
6425 }),
6426 })
6427 }
6428
6429 fn symbol_signature(&self, project_path: &ProjectPath) -> [u8; 32] {
6430 let mut hasher = Sha256::new();
6431 hasher.update(project_path.worktree_id.to_proto().to_be_bytes());
6432 hasher.update(project_path.path.to_string_lossy().as_bytes());
6433 hasher.update(self.nonce.to_be_bytes());
6434 hasher.finalize().as_slice().try_into().unwrap()
6435 }
6436
6437 async fn handle_open_buffer_by_id(
6438 this: ModelHandle<Self>,
6439 envelope: TypedEnvelope<proto::OpenBufferById>,
6440 _: Arc<Client>,
6441 mut cx: AsyncAppContext,
6442 ) -> Result<proto::OpenBufferResponse> {
6443 let peer_id = envelope.original_sender_id()?;
6444 let buffer = this
6445 .update(&mut cx, |this, cx| {
6446 this.open_buffer_by_id(envelope.payload.id, cx)
6447 })
6448 .await?;
6449 this.update(&mut cx, |this, cx| {
6450 Ok(proto::OpenBufferResponse {
6451 buffer_id: this.create_buffer_for_peer(&buffer, peer_id, cx),
6452 })
6453 })
6454 }
6455
6456 async fn handle_open_buffer_by_path(
6457 this: ModelHandle<Self>,
6458 envelope: TypedEnvelope<proto::OpenBufferByPath>,
6459 _: Arc<Client>,
6460 mut cx: AsyncAppContext,
6461 ) -> Result<proto::OpenBufferResponse> {
6462 let peer_id = envelope.original_sender_id()?;
6463 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
6464 let open_buffer = this.update(&mut cx, |this, cx| {
6465 this.open_buffer(
6466 ProjectPath {
6467 worktree_id,
6468 path: PathBuf::from(envelope.payload.path).into(),
6469 },
6470 cx,
6471 )
6472 });
6473
6474 let buffer = open_buffer.await?;
6475 this.update(&mut cx, |this, cx| {
6476 Ok(proto::OpenBufferResponse {
6477 buffer_id: this.create_buffer_for_peer(&buffer, peer_id, cx),
6478 })
6479 })
6480 }
6481
6482 fn serialize_project_transaction_for_peer(
6483 &mut self,
6484 project_transaction: ProjectTransaction,
6485 peer_id: proto::PeerId,
6486 cx: &mut AppContext,
6487 ) -> proto::ProjectTransaction {
6488 let mut serialized_transaction = proto::ProjectTransaction {
6489 buffer_ids: Default::default(),
6490 transactions: Default::default(),
6491 };
6492 for (buffer, transaction) in project_transaction.0 {
6493 serialized_transaction
6494 .buffer_ids
6495 .push(self.create_buffer_for_peer(&buffer, peer_id, cx));
6496 serialized_transaction
6497 .transactions
6498 .push(language::proto::serialize_transaction(&transaction));
6499 }
6500 serialized_transaction
6501 }
6502
6503 fn deserialize_project_transaction(
6504 &mut self,
6505 message: proto::ProjectTransaction,
6506 push_to_history: bool,
6507 cx: &mut ModelContext<Self>,
6508 ) -> Task<Result<ProjectTransaction>> {
6509 cx.spawn(|this, mut cx| async move {
6510 let mut project_transaction = ProjectTransaction::default();
6511 for (buffer_id, transaction) in message.buffer_ids.into_iter().zip(message.transactions)
6512 {
6513 let buffer = this
6514 .update(&mut cx, |this, cx| {
6515 this.wait_for_remote_buffer(buffer_id, cx)
6516 })
6517 .await?;
6518 let transaction = language::proto::deserialize_transaction(transaction)?;
6519 project_transaction.0.insert(buffer, transaction);
6520 }
6521
6522 for (buffer, transaction) in &project_transaction.0 {
6523 buffer
6524 .update(&mut cx, |buffer, _| {
6525 buffer.wait_for_edits(transaction.edit_ids.iter().copied())
6526 })
6527 .await?;
6528
6529 if push_to_history {
6530 buffer.update(&mut cx, |buffer, _| {
6531 buffer.push_transaction(transaction.clone(), Instant::now());
6532 });
6533 }
6534 }
6535
6536 Ok(project_transaction)
6537 })
6538 }
6539
6540 fn create_buffer_for_peer(
6541 &mut self,
6542 buffer: &ModelHandle<Buffer>,
6543 peer_id: proto::PeerId,
6544 cx: &mut AppContext,
6545 ) -> u64 {
6546 let buffer_id = buffer.read(cx).remote_id();
6547 if let Some(ProjectClientState::Local { updates_tx, .. }) = &self.client_state {
6548 updates_tx
6549 .unbounded_send(LocalProjectUpdate::CreateBufferForPeer { peer_id, buffer_id })
6550 .ok();
6551 }
6552 buffer_id
6553 }
6554
6555 fn wait_for_remote_buffer(
6556 &mut self,
6557 id: u64,
6558 cx: &mut ModelContext<Self>,
6559 ) -> Task<Result<ModelHandle<Buffer>>> {
6560 let mut opened_buffer_rx = self.opened_buffer.1.clone();
6561
6562 cx.spawn_weak(|this, mut cx| async move {
6563 let buffer = loop {
6564 let Some(this) = this.upgrade(&cx) else {
6565 return Err(anyhow!("project dropped"));
6566 };
6567
6568 let buffer = this.read_with(&cx, |this, cx| {
6569 this.opened_buffers
6570 .get(&id)
6571 .and_then(|buffer| buffer.upgrade(cx))
6572 });
6573
6574 if let Some(buffer) = buffer {
6575 break buffer;
6576 } else if this.read_with(&cx, |this, _| this.is_read_only()) {
6577 return Err(anyhow!("disconnected before buffer {} could be opened", id));
6578 }
6579
6580 this.update(&mut cx, |this, _| {
6581 this.incomplete_remote_buffers.entry(id).or_default();
6582 });
6583 drop(this);
6584
6585 opened_buffer_rx
6586 .next()
6587 .await
6588 .ok_or_else(|| anyhow!("project dropped while waiting for buffer"))?;
6589 };
6590
6591 Ok(buffer)
6592 })
6593 }
6594
6595 fn synchronize_remote_buffers(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
6596 let project_id = match self.client_state.as_ref() {
6597 Some(ProjectClientState::Remote {
6598 sharing_has_stopped,
6599 remote_id,
6600 ..
6601 }) => {
6602 if *sharing_has_stopped {
6603 return Task::ready(Err(anyhow!(
6604 "can't synchronize remote buffers on a readonly project"
6605 )));
6606 } else {
6607 *remote_id
6608 }
6609 }
6610 Some(ProjectClientState::Local { .. }) | None => {
6611 return Task::ready(Err(anyhow!(
6612 "can't synchronize remote buffers on a local project"
6613 )))
6614 }
6615 };
6616
6617 let client = self.client.clone();
6618 cx.spawn(|this, cx| async move {
6619 let (buffers, incomplete_buffer_ids) = this.read_with(&cx, |this, cx| {
6620 let buffers = this
6621 .opened_buffers
6622 .iter()
6623 .filter_map(|(id, buffer)| {
6624 let buffer = buffer.upgrade(cx)?;
6625 Some(proto::BufferVersion {
6626 id: *id,
6627 version: language::proto::serialize_version(&buffer.read(cx).version),
6628 })
6629 })
6630 .collect();
6631 let incomplete_buffer_ids = this
6632 .incomplete_remote_buffers
6633 .keys()
6634 .copied()
6635 .collect::<Vec<_>>();
6636
6637 (buffers, incomplete_buffer_ids)
6638 });
6639 let response = client
6640 .request(proto::SynchronizeBuffers {
6641 project_id,
6642 buffers,
6643 })
6644 .await?;
6645
6646 let send_updates_for_buffers = response.buffers.into_iter().map(|buffer| {
6647 let client = client.clone();
6648 let buffer_id = buffer.id;
6649 let remote_version = language::proto::deserialize_version(&buffer.version);
6650 this.read_with(&cx, |this, cx| {
6651 if let Some(buffer) = this.buffer_for_id(buffer_id, cx) {
6652 let operations = buffer.read(cx).serialize_ops(Some(remote_version), cx);
6653 cx.background().spawn(async move {
6654 let operations = operations.await;
6655 for chunk in split_operations(operations) {
6656 client
6657 .request(proto::UpdateBuffer {
6658 project_id,
6659 buffer_id,
6660 operations: chunk,
6661 })
6662 .await?;
6663 }
6664 anyhow::Ok(())
6665 })
6666 } else {
6667 Task::ready(Ok(()))
6668 }
6669 })
6670 });
6671
6672 // Any incomplete buffers have open requests waiting. Request that the host sends
6673 // creates these buffers for us again to unblock any waiting futures.
6674 for id in incomplete_buffer_ids {
6675 cx.background()
6676 .spawn(client.request(proto::OpenBufferById { project_id, id }))
6677 .detach();
6678 }
6679
6680 futures::future::join_all(send_updates_for_buffers)
6681 .await
6682 .into_iter()
6683 .collect()
6684 })
6685 }
6686
6687 pub fn worktree_metadata_protos(&self, cx: &AppContext) -> Vec<proto::WorktreeMetadata> {
6688 self.worktrees(cx)
6689 .map(|worktree| {
6690 let worktree = worktree.read(cx);
6691 proto::WorktreeMetadata {
6692 id: worktree.id().to_proto(),
6693 root_name: worktree.root_name().into(),
6694 visible: worktree.is_visible(),
6695 abs_path: worktree.abs_path().to_string_lossy().into(),
6696 }
6697 })
6698 .collect()
6699 }
6700
6701 fn set_worktrees_from_proto(
6702 &mut self,
6703 worktrees: Vec<proto::WorktreeMetadata>,
6704 cx: &mut ModelContext<Project>,
6705 ) -> Result<()> {
6706 let replica_id = self.replica_id();
6707 let remote_id = self.remote_id().ok_or_else(|| anyhow!("invalid project"))?;
6708
6709 let mut old_worktrees_by_id = self
6710 .worktrees
6711 .drain(..)
6712 .filter_map(|worktree| {
6713 let worktree = worktree.upgrade(cx)?;
6714 Some((worktree.read(cx).id(), worktree))
6715 })
6716 .collect::<HashMap<_, _>>();
6717
6718 for worktree in worktrees {
6719 if let Some(old_worktree) =
6720 old_worktrees_by_id.remove(&WorktreeId::from_proto(worktree.id))
6721 {
6722 self.worktrees.push(WorktreeHandle::Strong(old_worktree));
6723 } else {
6724 let worktree =
6725 Worktree::remote(remote_id, replica_id, worktree, self.client.clone(), cx);
6726 let _ = self.add_worktree(&worktree, cx);
6727 }
6728 }
6729
6730 self.metadata_changed(cx);
6731 for id in old_worktrees_by_id.keys() {
6732 cx.emit(Event::WorktreeRemoved(*id));
6733 }
6734
6735 Ok(())
6736 }
6737
6738 fn set_collaborators_from_proto(
6739 &mut self,
6740 messages: Vec<proto::Collaborator>,
6741 cx: &mut ModelContext<Self>,
6742 ) -> Result<()> {
6743 let mut collaborators = HashMap::default();
6744 for message in messages {
6745 let collaborator = Collaborator::from_proto(message)?;
6746 collaborators.insert(collaborator.peer_id, collaborator);
6747 }
6748 for old_peer_id in self.collaborators.keys() {
6749 if !collaborators.contains_key(old_peer_id) {
6750 cx.emit(Event::CollaboratorLeft(*old_peer_id));
6751 }
6752 }
6753 self.collaborators = collaborators;
6754 Ok(())
6755 }
6756
6757 fn deserialize_symbol(
6758 &self,
6759 serialized_symbol: proto::Symbol,
6760 ) -> impl Future<Output = Result<Symbol>> {
6761 let languages = self.languages.clone();
6762 async move {
6763 let source_worktree_id = WorktreeId::from_proto(serialized_symbol.source_worktree_id);
6764 let worktree_id = WorktreeId::from_proto(serialized_symbol.worktree_id);
6765 let start = serialized_symbol
6766 .start
6767 .ok_or_else(|| anyhow!("invalid start"))?;
6768 let end = serialized_symbol
6769 .end
6770 .ok_or_else(|| anyhow!("invalid end"))?;
6771 let kind = unsafe { mem::transmute(serialized_symbol.kind) };
6772 let path = ProjectPath {
6773 worktree_id,
6774 path: PathBuf::from(serialized_symbol.path).into(),
6775 };
6776 let language = languages
6777 .language_for_file(&path.path, None)
6778 .await
6779 .log_err();
6780 Ok(Symbol {
6781 language_server_name: LanguageServerName(
6782 serialized_symbol.language_server_name.into(),
6783 ),
6784 source_worktree_id,
6785 path,
6786 label: {
6787 match language {
6788 Some(language) => {
6789 language
6790 .label_for_symbol(&serialized_symbol.name, kind)
6791 .await
6792 }
6793 None => None,
6794 }
6795 .unwrap_or_else(|| CodeLabel::plain(serialized_symbol.name.clone(), None))
6796 },
6797
6798 name: serialized_symbol.name,
6799 range: Unclipped(PointUtf16::new(start.row, start.column))
6800 ..Unclipped(PointUtf16::new(end.row, end.column)),
6801 kind,
6802 signature: serialized_symbol
6803 .signature
6804 .try_into()
6805 .map_err(|_| anyhow!("invalid signature"))?,
6806 })
6807 }
6808 }
6809
6810 async fn handle_buffer_saved(
6811 this: ModelHandle<Self>,
6812 envelope: TypedEnvelope<proto::BufferSaved>,
6813 _: Arc<Client>,
6814 mut cx: AsyncAppContext,
6815 ) -> Result<()> {
6816 let fingerprint = deserialize_fingerprint(&envelope.payload.fingerprint)?;
6817 let version = deserialize_version(&envelope.payload.version);
6818 let mtime = envelope
6819 .payload
6820 .mtime
6821 .ok_or_else(|| anyhow!("missing mtime"))?
6822 .into();
6823
6824 this.update(&mut cx, |this, cx| {
6825 let buffer = this
6826 .opened_buffers
6827 .get(&envelope.payload.buffer_id)
6828 .and_then(|buffer| buffer.upgrade(cx))
6829 .or_else(|| {
6830 this.incomplete_remote_buffers
6831 .get(&envelope.payload.buffer_id)
6832 .and_then(|b| b.clone())
6833 });
6834 if let Some(buffer) = buffer {
6835 buffer.update(cx, |buffer, cx| {
6836 buffer.did_save(version, fingerprint, mtime, cx);
6837 });
6838 }
6839 Ok(())
6840 })
6841 }
6842
6843 async fn handle_buffer_reloaded(
6844 this: ModelHandle<Self>,
6845 envelope: TypedEnvelope<proto::BufferReloaded>,
6846 _: Arc<Client>,
6847 mut cx: AsyncAppContext,
6848 ) -> Result<()> {
6849 let payload = envelope.payload;
6850 let version = deserialize_version(&payload.version);
6851 let fingerprint = deserialize_fingerprint(&payload.fingerprint)?;
6852 let line_ending = deserialize_line_ending(
6853 proto::LineEnding::from_i32(payload.line_ending)
6854 .ok_or_else(|| anyhow!("missing line ending"))?,
6855 );
6856 let mtime = payload
6857 .mtime
6858 .ok_or_else(|| anyhow!("missing mtime"))?
6859 .into();
6860 this.update(&mut cx, |this, cx| {
6861 let buffer = this
6862 .opened_buffers
6863 .get(&payload.buffer_id)
6864 .and_then(|buffer| buffer.upgrade(cx))
6865 .or_else(|| {
6866 this.incomplete_remote_buffers
6867 .get(&payload.buffer_id)
6868 .cloned()
6869 .flatten()
6870 });
6871 if let Some(buffer) = buffer {
6872 buffer.update(cx, |buffer, cx| {
6873 buffer.did_reload(version, fingerprint, line_ending, mtime, cx);
6874 });
6875 }
6876 Ok(())
6877 })
6878 }
6879
6880 #[allow(clippy::type_complexity)]
6881 fn edits_from_lsp(
6882 &mut self,
6883 buffer: &ModelHandle<Buffer>,
6884 lsp_edits: impl 'static + Send + IntoIterator<Item = lsp::TextEdit>,
6885 server_id: LanguageServerId,
6886 version: Option<i32>,
6887 cx: &mut ModelContext<Self>,
6888 ) -> Task<Result<Vec<(Range<Anchor>, String)>>> {
6889 let snapshot = self.buffer_snapshot_for_lsp_version(buffer, server_id, version, cx);
6890 cx.background().spawn(async move {
6891 let snapshot = snapshot?;
6892 let mut lsp_edits = lsp_edits
6893 .into_iter()
6894 .map(|edit| (range_from_lsp(edit.range), edit.new_text))
6895 .collect::<Vec<_>>();
6896 lsp_edits.sort_by_key(|(range, _)| range.start);
6897
6898 let mut lsp_edits = lsp_edits.into_iter().peekable();
6899 let mut edits = Vec::new();
6900 while let Some((range, mut new_text)) = lsp_edits.next() {
6901 // Clip invalid ranges provided by the language server.
6902 let mut range = snapshot.clip_point_utf16(range.start, Bias::Left)
6903 ..snapshot.clip_point_utf16(range.end, Bias::Left);
6904
6905 // Combine any LSP edits that are adjacent.
6906 //
6907 // Also, combine LSP edits that are separated from each other by only
6908 // a newline. This is important because for some code actions,
6909 // Rust-analyzer rewrites the entire buffer via a series of edits that
6910 // are separated by unchanged newline characters.
6911 //
6912 // In order for the diffing logic below to work properly, any edits that
6913 // cancel each other out must be combined into one.
6914 while let Some((next_range, next_text)) = lsp_edits.peek() {
6915 if next_range.start.0 > range.end {
6916 if next_range.start.0.row > range.end.row + 1
6917 || next_range.start.0.column > 0
6918 || snapshot.clip_point_utf16(
6919 Unclipped(PointUtf16::new(range.end.row, u32::MAX)),
6920 Bias::Left,
6921 ) > range.end
6922 {
6923 break;
6924 }
6925 new_text.push('\n');
6926 }
6927 range.end = snapshot.clip_point_utf16(next_range.end, Bias::Left);
6928 new_text.push_str(next_text);
6929 lsp_edits.next();
6930 }
6931
6932 // For multiline edits, perform a diff of the old and new text so that
6933 // we can identify the changes more precisely, preserving the locations
6934 // of any anchors positioned in the unchanged regions.
6935 if range.end.row > range.start.row {
6936 let mut offset = range.start.to_offset(&snapshot);
6937 let old_text = snapshot.text_for_range(range).collect::<String>();
6938
6939 let diff = TextDiff::from_lines(old_text.as_str(), &new_text);
6940 let mut moved_since_edit = true;
6941 for change in diff.iter_all_changes() {
6942 let tag = change.tag();
6943 let value = change.value();
6944 match tag {
6945 ChangeTag::Equal => {
6946 offset += value.len();
6947 moved_since_edit = true;
6948 }
6949 ChangeTag::Delete => {
6950 let start = snapshot.anchor_after(offset);
6951 let end = snapshot.anchor_before(offset + value.len());
6952 if moved_since_edit {
6953 edits.push((start..end, String::new()));
6954 } else {
6955 edits.last_mut().unwrap().0.end = end;
6956 }
6957 offset += value.len();
6958 moved_since_edit = false;
6959 }
6960 ChangeTag::Insert => {
6961 if moved_since_edit {
6962 let anchor = snapshot.anchor_after(offset);
6963 edits.push((anchor..anchor, value.to_string()));
6964 } else {
6965 edits.last_mut().unwrap().1.push_str(value);
6966 }
6967 moved_since_edit = false;
6968 }
6969 }
6970 }
6971 } else if range.end == range.start {
6972 let anchor = snapshot.anchor_after(range.start);
6973 edits.push((anchor..anchor, new_text));
6974 } else {
6975 let edit_start = snapshot.anchor_after(range.start);
6976 let edit_end = snapshot.anchor_before(range.end);
6977 edits.push((edit_start..edit_end, new_text));
6978 }
6979 }
6980
6981 Ok(edits)
6982 })
6983 }
6984
6985 fn buffer_snapshot_for_lsp_version(
6986 &mut self,
6987 buffer: &ModelHandle<Buffer>,
6988 server_id: LanguageServerId,
6989 version: Option<i32>,
6990 cx: &AppContext,
6991 ) -> Result<TextBufferSnapshot> {
6992 const OLD_VERSIONS_TO_RETAIN: i32 = 10;
6993
6994 if let Some(version) = version {
6995 let buffer_id = buffer.read(cx).remote_id();
6996 let snapshots = self
6997 .buffer_snapshots
6998 .get_mut(&buffer_id)
6999 .and_then(|m| m.get_mut(&server_id))
7000 .ok_or_else(|| {
7001 anyhow!("no snapshots found for buffer {buffer_id} and server {server_id}")
7002 })?;
7003
7004 let found_snapshot = snapshots
7005 .binary_search_by_key(&version, |e| e.version)
7006 .map(|ix| snapshots[ix].snapshot.clone())
7007 .map_err(|_| {
7008 anyhow!("snapshot not found for buffer {buffer_id} server {server_id} at version {version}")
7009 })?;
7010
7011 snapshots.retain(|snapshot| snapshot.version + OLD_VERSIONS_TO_RETAIN >= version);
7012 Ok(found_snapshot)
7013 } else {
7014 Ok((buffer.read(cx)).text_snapshot())
7015 }
7016 }
7017
7018 pub fn language_servers(
7019 &self,
7020 ) -> impl '_ + Iterator<Item = (LanguageServerId, LanguageServerName, WorktreeId)> {
7021 self.language_server_ids
7022 .iter()
7023 .map(|((worktree_id, server_name), server_id)| {
7024 (*server_id, server_name.clone(), *worktree_id)
7025 })
7026 }
7027
7028 pub fn language_server_for_id(&self, id: LanguageServerId) -> Option<Arc<LanguageServer>> {
7029 if let LanguageServerState::Running { server, .. } = self.language_servers.get(&id)? {
7030 Some(server.clone())
7031 } else {
7032 None
7033 }
7034 }
7035
7036 pub fn language_servers_for_buffer(
7037 &self,
7038 buffer: &Buffer,
7039 cx: &AppContext,
7040 ) -> impl Iterator<Item = (&Arc<CachedLspAdapter>, &Arc<LanguageServer>)> {
7041 self.language_server_ids_for_buffer(buffer, cx)
7042 .into_iter()
7043 .filter_map(|server_id| {
7044 let server = self.language_servers.get(&server_id)?;
7045 if let LanguageServerState::Running {
7046 adapter, server, ..
7047 } = server
7048 {
7049 Some((adapter, server))
7050 } else {
7051 None
7052 }
7053 })
7054 }
7055
7056 fn primary_language_servers_for_buffer(
7057 &self,
7058 buffer: &Buffer,
7059 cx: &AppContext,
7060 ) -> Option<(&Arc<CachedLspAdapter>, &Arc<LanguageServer>)> {
7061 self.language_servers_for_buffer(buffer, cx).next()
7062 }
7063
7064 fn language_server_for_buffer(
7065 &self,
7066 buffer: &Buffer,
7067 server_id: LanguageServerId,
7068 cx: &AppContext,
7069 ) -> Option<(&Arc<CachedLspAdapter>, &Arc<LanguageServer>)> {
7070 self.language_servers_for_buffer(buffer, cx)
7071 .find(|(_, s)| s.server_id() == server_id)
7072 }
7073
7074 fn language_server_ids_for_buffer(
7075 &self,
7076 buffer: &Buffer,
7077 cx: &AppContext,
7078 ) -> Vec<LanguageServerId> {
7079 if let Some((file, language)) = File::from_dyn(buffer.file()).zip(buffer.language()) {
7080 let worktree_id = file.worktree_id(cx);
7081 language
7082 .lsp_adapters()
7083 .iter()
7084 .flat_map(|adapter| {
7085 let key = (worktree_id, adapter.name.clone());
7086 self.language_server_ids.get(&key).copied()
7087 })
7088 .collect()
7089 } else {
7090 Vec::new()
7091 }
7092 }
7093}
7094
7095impl WorktreeHandle {
7096 pub fn upgrade(&self, cx: &AppContext) -> Option<ModelHandle<Worktree>> {
7097 match self {
7098 WorktreeHandle::Strong(handle) => Some(handle.clone()),
7099 WorktreeHandle::Weak(handle) => handle.upgrade(cx),
7100 }
7101 }
7102
7103 pub fn handle_id(&self) -> usize {
7104 match self {
7105 WorktreeHandle::Strong(handle) => handle.id(),
7106 WorktreeHandle::Weak(handle) => handle.id(),
7107 }
7108 }
7109}
7110
7111impl OpenBuffer {
7112 pub fn upgrade(&self, cx: &impl BorrowAppContext) -> Option<ModelHandle<Buffer>> {
7113 match self {
7114 OpenBuffer::Strong(handle) => Some(handle.clone()),
7115 OpenBuffer::Weak(handle) => handle.upgrade(cx),
7116 OpenBuffer::Operations(_) => None,
7117 }
7118 }
7119}
7120
7121pub struct PathMatchCandidateSet {
7122 pub snapshot: Snapshot,
7123 pub include_ignored: bool,
7124 pub include_root_name: bool,
7125}
7126
7127impl<'a> fuzzy::PathMatchCandidateSet<'a> for PathMatchCandidateSet {
7128 type Candidates = PathMatchCandidateSetIter<'a>;
7129
7130 fn id(&self) -> usize {
7131 self.snapshot.id().to_usize()
7132 }
7133
7134 fn len(&self) -> usize {
7135 if self.include_ignored {
7136 self.snapshot.file_count()
7137 } else {
7138 self.snapshot.visible_file_count()
7139 }
7140 }
7141
7142 fn prefix(&self) -> Arc<str> {
7143 if self.snapshot.root_entry().map_or(false, |e| e.is_file()) {
7144 self.snapshot.root_name().into()
7145 } else if self.include_root_name {
7146 format!("{}/", self.snapshot.root_name()).into()
7147 } else {
7148 "".into()
7149 }
7150 }
7151
7152 fn candidates(&'a self, start: usize) -> Self::Candidates {
7153 PathMatchCandidateSetIter {
7154 traversal: self.snapshot.files(self.include_ignored, start),
7155 }
7156 }
7157}
7158
7159pub struct PathMatchCandidateSetIter<'a> {
7160 traversal: Traversal<'a>,
7161}
7162
7163impl<'a> Iterator for PathMatchCandidateSetIter<'a> {
7164 type Item = fuzzy::PathMatchCandidate<'a>;
7165
7166 fn next(&mut self) -> Option<Self::Item> {
7167 self.traversal.next().map(|entry| {
7168 if let EntryKind::File(char_bag) = entry.kind {
7169 fuzzy::PathMatchCandidate {
7170 path: &entry.path,
7171 char_bag,
7172 }
7173 } else {
7174 unreachable!()
7175 }
7176 })
7177 }
7178}
7179
7180impl Entity for Project {
7181 type Event = Event;
7182
7183 fn release(&mut self, cx: &mut gpui::AppContext) {
7184 match &self.client_state {
7185 Some(ProjectClientState::Local { .. }) => {
7186 let _ = self.unshare_internal(cx);
7187 }
7188 Some(ProjectClientState::Remote { remote_id, .. }) => {
7189 let _ = self.client.send(proto::LeaveProject {
7190 project_id: *remote_id,
7191 });
7192 self.disconnected_from_host_internal(cx);
7193 }
7194 _ => {}
7195 }
7196 }
7197
7198 fn app_will_quit(
7199 &mut self,
7200 _: &mut AppContext,
7201 ) -> Option<std::pin::Pin<Box<dyn 'static + Future<Output = ()>>>> {
7202 let shutdown_futures = self
7203 .language_servers
7204 .drain()
7205 .map(|(_, server_state)| async {
7206 match server_state {
7207 LanguageServerState::Running { server, .. } => server.shutdown()?.await,
7208 LanguageServerState::Starting(starting_server) => {
7209 starting_server.await?.shutdown()?.await
7210 }
7211 }
7212 })
7213 .collect::<Vec<_>>();
7214
7215 Some(
7216 async move {
7217 futures::future::join_all(shutdown_futures).await;
7218 }
7219 .boxed(),
7220 )
7221 }
7222}
7223
7224impl Collaborator {
7225 fn from_proto(message: proto::Collaborator) -> Result<Self> {
7226 Ok(Self {
7227 peer_id: message.peer_id.ok_or_else(|| anyhow!("invalid peer id"))?,
7228 replica_id: message.replica_id as ReplicaId,
7229 })
7230 }
7231}
7232
7233impl<P: AsRef<Path>> From<(WorktreeId, P)> for ProjectPath {
7234 fn from((worktree_id, path): (WorktreeId, P)) -> Self {
7235 Self {
7236 worktree_id,
7237 path: path.as_ref().into(),
7238 }
7239 }
7240}
7241
7242impl ProjectLspAdapterDelegate {
7243 fn new(project: &Project, cx: &ModelContext<Project>) -> Arc<Self> {
7244 Arc::new(Self {
7245 project: cx.handle(),
7246 http_client: project.client.http_client(),
7247 })
7248 }
7249}
7250
7251impl LspAdapterDelegate for ProjectLspAdapterDelegate {
7252 fn show_notification(&self, message: &str, cx: &mut AppContext) {
7253 self.project
7254 .update(cx, |_, cx| cx.emit(Event::Notification(message.to_owned())));
7255 }
7256
7257 fn http_client(&self) -> Arc<dyn HttpClient> {
7258 self.http_client.clone()
7259 }
7260}
7261
7262fn split_operations(
7263 mut operations: Vec<proto::Operation>,
7264) -> impl Iterator<Item = Vec<proto::Operation>> {
7265 #[cfg(any(test, feature = "test-support"))]
7266 const CHUNK_SIZE: usize = 5;
7267
7268 #[cfg(not(any(test, feature = "test-support")))]
7269 const CHUNK_SIZE: usize = 100;
7270
7271 let mut done = false;
7272 std::iter::from_fn(move || {
7273 if done {
7274 return None;
7275 }
7276
7277 let operations = operations
7278 .drain(..cmp::min(CHUNK_SIZE, operations.len()))
7279 .collect::<Vec<_>>();
7280 if operations.is_empty() {
7281 done = true;
7282 }
7283 Some(operations)
7284 })
7285}
7286
7287fn serialize_symbol(symbol: &Symbol) -> proto::Symbol {
7288 proto::Symbol {
7289 language_server_name: symbol.language_server_name.0.to_string(),
7290 source_worktree_id: symbol.source_worktree_id.to_proto(),
7291 worktree_id: symbol.path.worktree_id.to_proto(),
7292 path: symbol.path.path.to_string_lossy().to_string(),
7293 name: symbol.name.clone(),
7294 kind: unsafe { mem::transmute(symbol.kind) },
7295 start: Some(proto::PointUtf16 {
7296 row: symbol.range.start.0.row,
7297 column: symbol.range.start.0.column,
7298 }),
7299 end: Some(proto::PointUtf16 {
7300 row: symbol.range.end.0.row,
7301 column: symbol.range.end.0.column,
7302 }),
7303 signature: symbol.signature.to_vec(),
7304 }
7305}
7306
7307fn relativize_path(base: &Path, path: &Path) -> PathBuf {
7308 let mut path_components = path.components();
7309 let mut base_components = base.components();
7310 let mut components: Vec<Component> = Vec::new();
7311 loop {
7312 match (path_components.next(), base_components.next()) {
7313 (None, None) => break,
7314 (Some(a), None) => {
7315 components.push(a);
7316 components.extend(path_components.by_ref());
7317 break;
7318 }
7319 (None, _) => components.push(Component::ParentDir),
7320 (Some(a), Some(b)) if components.is_empty() && a == b => (),
7321 (Some(a), Some(b)) if b == Component::CurDir => components.push(a),
7322 (Some(a), Some(_)) => {
7323 components.push(Component::ParentDir);
7324 for _ in base_components {
7325 components.push(Component::ParentDir);
7326 }
7327 components.push(a);
7328 components.extend(path_components.by_ref());
7329 break;
7330 }
7331 }
7332 }
7333 components.iter().map(|c| c.as_os_str()).collect()
7334}
7335
7336impl Item for Buffer {
7337 fn entry_id(&self, cx: &AppContext) -> Option<ProjectEntryId> {
7338 File::from_dyn(self.file()).and_then(|file| file.project_entry_id(cx))
7339 }
7340
7341 fn project_path(&self, cx: &AppContext) -> Option<ProjectPath> {
7342 File::from_dyn(self.file()).map(|file| ProjectPath {
7343 worktree_id: file.worktree_id(cx),
7344 path: file.path().clone(),
7345 })
7346 }
7347}
7348
7349async fn wait_for_loading_buffer(
7350 mut receiver: postage::watch::Receiver<Option<Result<ModelHandle<Buffer>, Arc<anyhow::Error>>>>,
7351) -> Result<ModelHandle<Buffer>, Arc<anyhow::Error>> {
7352 loop {
7353 if let Some(result) = receiver.borrow().as_ref() {
7354 match result {
7355 Ok(buffer) => return Ok(buffer.to_owned()),
7356 Err(e) => return Err(e.to_owned()),
7357 }
7358 }
7359 receiver.next().await;
7360 }
7361}