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