1use crate::{
2 ProjectItem as _, ProjectPath,
3 lsp_store::OpenLspBufferHandle,
4 search::SearchQuery,
5 worktree_store::{WorktreeStore, WorktreeStoreEvent},
6};
7use anyhow::{Context as _, Result, anyhow};
8use client::Client;
9use collections::{HashMap, HashSet, hash_map};
10use encodings::Encoding;
11use futures::{Future, FutureExt as _, channel::oneshot, future::Shared};
12use gpui::{
13 App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Subscription, Task, WeakEntity,
14};
15use language::{
16 Buffer, BufferEvent, Capability, DiskState, File as _, Language, Operation,
17 proto::{
18 deserialize_line_ending, deserialize_version, serialize_line_ending, serialize_version,
19 split_operations,
20 },
21};
22use rpc::{
23 AnyProtoClient, ErrorCode, ErrorExt as _, TypedEnvelope,
24 proto::{self},
25};
26use smol::channel::Receiver;
27use std::{io, pin::pin, sync::Arc, time::Instant};
28use text::{BufferId, ReplicaId};
29use util::{ResultExt as _, TryFutureExt, debug_panic, maybe, rel_path::RelPath};
30use worktree::{File, PathChange, ProjectEntryId, Worktree, WorktreeId};
31
32/// A set of open buffers.
33pub struct BufferStore {
34 state: BufferStoreState,
35 #[allow(clippy::type_complexity)]
36 loading_buffers: HashMap<ProjectPath, Shared<Task<Result<Entity<Buffer>, Arc<anyhow::Error>>>>>,
37 worktree_store: Entity<WorktreeStore>,
38 opened_buffers: HashMap<BufferId, OpenBuffer>,
39 path_to_buffer_id: HashMap<ProjectPath, BufferId>,
40 downstream_client: Option<(AnyProtoClient, u64)>,
41 shared_buffers: HashMap<proto::PeerId, HashMap<BufferId, SharedBuffer>>,
42 non_searchable_buffers: HashSet<BufferId>,
43}
44
45#[derive(Hash, Eq, PartialEq, Clone)]
46struct SharedBuffer {
47 buffer: Entity<Buffer>,
48 lsp_handle: Option<OpenLspBufferHandle>,
49}
50
51enum BufferStoreState {
52 Local(LocalBufferStore),
53 Remote(RemoteBufferStore),
54}
55
56struct RemoteBufferStore {
57 shared_with_me: HashSet<Entity<Buffer>>,
58 upstream_client: AnyProtoClient,
59 project_id: u64,
60 loading_remote_buffers_by_id: HashMap<BufferId, Entity<Buffer>>,
61 remote_buffer_listeners:
62 HashMap<BufferId, Vec<oneshot::Sender<anyhow::Result<Entity<Buffer>>>>>,
63 worktree_store: Entity<WorktreeStore>,
64}
65
66struct LocalBufferStore {
67 local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,
68 worktree_store: Entity<WorktreeStore>,
69 _subscription: Subscription,
70}
71
72enum OpenBuffer {
73 Complete { buffer: WeakEntity<Buffer> },
74 Operations(Vec<Operation>),
75}
76
77pub enum BufferStoreEvent {
78 BufferAdded(Entity<Buffer>),
79 BufferOpened {
80 buffer: Entity<Buffer>,
81 project_path: ProjectPath,
82 },
83 SharedBufferClosed(proto::PeerId, BufferId),
84 BufferDropped(BufferId),
85 BufferChangedFilePath {
86 buffer: Entity<Buffer>,
87 old_file: Option<Arc<dyn language::File>>,
88 },
89}
90
91#[derive(Default, Debug, Clone)]
92pub struct ProjectTransaction(pub HashMap<Entity<Buffer>, language::Transaction>);
93
94impl PartialEq for ProjectTransaction {
95 fn eq(&self, other: &Self) -> bool {
96 self.0.len() == other.0.len()
97 && self.0.iter().all(|(buffer, transaction)| {
98 other.0.get(buffer).is_some_and(|t| t.id == transaction.id)
99 })
100 }
101}
102
103impl EventEmitter<BufferStoreEvent> for BufferStore {}
104
105impl RemoteBufferStore {
106 pub fn wait_for_remote_buffer(
107 &mut self,
108 id: BufferId,
109 cx: &mut Context<BufferStore>,
110 ) -> Task<Result<Entity<Buffer>>> {
111 let (tx, rx) = oneshot::channel();
112 self.remote_buffer_listeners.entry(id).or_default().push(tx);
113
114 cx.spawn(async move |this, cx| {
115 if let Some(buffer) = this
116 .read_with(cx, |buffer_store, _| buffer_store.get(id))
117 .ok()
118 .flatten()
119 {
120 return Ok(buffer);
121 }
122
123 cx.background_spawn(async move { rx.await? }).await
124 })
125 }
126
127 fn save_remote_buffer(
128 &self,
129 buffer_handle: Entity<Buffer>,
130 new_path: Option<proto::ProjectPath>,
131 cx: &Context<BufferStore>,
132 ) -> Task<Result<()>> {
133 let buffer = buffer_handle.read(cx);
134 let buffer_id = buffer.remote_id().into();
135 let version = buffer.version();
136 let rpc = self.upstream_client.clone();
137 let project_id = self.project_id;
138 cx.spawn(async move |_, cx| {
139 let response = rpc
140 .request(proto::SaveBuffer {
141 project_id,
142 buffer_id,
143 new_path,
144 version: serialize_version(&version),
145 })
146 .await?;
147 let version = deserialize_version(&response.version);
148 let mtime = response.mtime.map(|mtime| mtime.into());
149
150 buffer_handle.update(cx, |buffer, cx| {
151 buffer.did_save(version.clone(), mtime, cx);
152 })?;
153
154 Ok(())
155 })
156 }
157
158 pub fn handle_create_buffer_for_peer(
159 &mut self,
160 envelope: TypedEnvelope<proto::CreateBufferForPeer>,
161 replica_id: ReplicaId,
162 capability: Capability,
163 cx: &mut Context<BufferStore>,
164 ) -> Result<Option<Entity<Buffer>>> {
165 match envelope.payload.variant.context("missing variant")? {
166 proto::create_buffer_for_peer::Variant::State(mut state) => {
167 let buffer_id = BufferId::new(state.id)?;
168
169 let buffer_result = maybe!({
170 let mut buffer_file = None;
171 if let Some(file) = state.file.take() {
172 let worktree_id = worktree::WorktreeId::from_proto(file.worktree_id);
173 let worktree = self
174 .worktree_store
175 .read(cx)
176 .worktree_for_id(worktree_id, cx)
177 .with_context(|| {
178 format!("no worktree found for id {}", file.worktree_id)
179 })?;
180 buffer_file = Some(Arc::new(File::from_proto(file, worktree, cx)?)
181 as Arc<dyn language::File>);
182 }
183 Buffer::from_proto(
184 replica_id,
185 capability,
186 state,
187 buffer_file,
188 cx.background_executor(),
189 )
190 });
191
192 match buffer_result {
193 Ok(buffer) => {
194 let buffer = cx.new(|_| buffer);
195 self.loading_remote_buffers_by_id.insert(buffer_id, buffer);
196 }
197 Err(error) => {
198 if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {
199 for listener in listeners {
200 listener.send(Err(anyhow!(error.cloned()))).ok();
201 }
202 }
203 }
204 }
205 }
206 proto::create_buffer_for_peer::Variant::Chunk(chunk) => {
207 let buffer_id = BufferId::new(chunk.buffer_id)?;
208 let buffer = self
209 .loading_remote_buffers_by_id
210 .get(&buffer_id)
211 .cloned()
212 .with_context(|| {
213 format!(
214 "received chunk for buffer {} without initial state",
215 chunk.buffer_id
216 )
217 })?;
218
219 let result = maybe!({
220 let operations = chunk
221 .operations
222 .into_iter()
223 .map(language::proto::deserialize_operation)
224 .collect::<Result<Vec<_>>>()?;
225 buffer.update(cx, |buffer, cx| buffer.apply_ops(operations, cx));
226 anyhow::Ok(())
227 });
228
229 if let Err(error) = result {
230 self.loading_remote_buffers_by_id.remove(&buffer_id);
231 if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {
232 for listener in listeners {
233 listener.send(Err(error.cloned())).ok();
234 }
235 }
236 } else if chunk.is_last {
237 self.loading_remote_buffers_by_id.remove(&buffer_id);
238 if self.upstream_client.is_via_collab() {
239 // retain buffers sent by peers to avoid races.
240 self.shared_with_me.insert(buffer.clone());
241 }
242
243 if let Some(senders) = self.remote_buffer_listeners.remove(&buffer_id) {
244 for sender in senders {
245 sender.send(Ok(buffer.clone())).ok();
246 }
247 }
248 return Ok(Some(buffer));
249 }
250 }
251 }
252 Ok(None)
253 }
254
255 pub fn incomplete_buffer_ids(&self) -> Vec<BufferId> {
256 self.loading_remote_buffers_by_id
257 .keys()
258 .copied()
259 .collect::<Vec<_>>()
260 }
261
262 pub fn deserialize_project_transaction(
263 &self,
264 message: proto::ProjectTransaction,
265 push_to_history: bool,
266 cx: &mut Context<BufferStore>,
267 ) -> Task<Result<ProjectTransaction>> {
268 cx.spawn(async move |this, cx| {
269 let mut project_transaction = ProjectTransaction::default();
270 for (buffer_id, transaction) in message.buffer_ids.into_iter().zip(message.transactions)
271 {
272 let buffer_id = BufferId::new(buffer_id)?;
273 let buffer = this
274 .update(cx, |this, cx| this.wait_for_remote_buffer(buffer_id, cx))?
275 .await?;
276 let transaction = language::proto::deserialize_transaction(transaction)?;
277 project_transaction.0.insert(buffer, transaction);
278 }
279
280 for (buffer, transaction) in &project_transaction.0 {
281 buffer
282 .update(cx, |buffer, _| {
283 buffer.wait_for_edits(transaction.edit_ids.iter().copied())
284 })?
285 .await?;
286
287 if push_to_history {
288 buffer.update(cx, |buffer, _| {
289 buffer.push_transaction(transaction.clone(), Instant::now());
290 buffer.finalize_last_transaction();
291 })?;
292 }
293 }
294
295 Ok(project_transaction)
296 })
297 }
298
299 fn open_buffer(
300 &self,
301 path: Arc<RelPath>,
302 worktree: Entity<Worktree>,
303 cx: &mut Context<BufferStore>,
304 ) -> Task<Result<Entity<Buffer>>> {
305 let worktree_id = worktree.read(cx).id().to_proto();
306 let project_id = self.project_id;
307 let client = self.upstream_client.clone();
308 cx.spawn(async move |this, cx| {
309 let response = client
310 .request(proto::OpenBufferByPath {
311 project_id,
312 worktree_id,
313 path: path.to_proto(),
314 })
315 .await?;
316 let buffer_id = BufferId::new(response.buffer_id)?;
317
318 let buffer = this
319 .update(cx, {
320 |this, cx| this.wait_for_remote_buffer(buffer_id, cx)
321 })?
322 .await?;
323
324 Ok(buffer)
325 })
326 }
327
328 fn create_buffer(
329 &self,
330 project_searchable: bool,
331 cx: &mut Context<BufferStore>,
332 ) -> Task<Result<Entity<Buffer>>> {
333 let create = self.upstream_client.request(proto::OpenNewBuffer {
334 project_id: self.project_id,
335 });
336 cx.spawn(async move |this, cx| {
337 let response = create.await?;
338 let buffer_id = BufferId::new(response.buffer_id)?;
339
340 this.update(cx, |this, cx| {
341 if !project_searchable {
342 this.non_searchable_buffers.insert(buffer_id);
343 }
344 this.wait_for_remote_buffer(buffer_id, cx)
345 })?
346 .await
347 })
348 }
349
350 fn reload_buffers(
351 &self,
352 buffers: HashSet<Entity<Buffer>>,
353 push_to_history: bool,
354 cx: &mut Context<BufferStore>,
355 ) -> Task<Result<ProjectTransaction>> {
356 let request = self.upstream_client.request(proto::ReloadBuffers {
357 project_id: self.project_id,
358 buffer_ids: buffers
359 .iter()
360 .map(|buffer| buffer.read(cx).remote_id().to_proto())
361 .collect(),
362 });
363
364 cx.spawn(async move |this, cx| {
365 let response = request.await?.transaction.context("missing transaction")?;
366 this.update(cx, |this, cx| {
367 this.deserialize_project_transaction(response, push_to_history, cx)
368 })?
369 .await
370 })
371 }
372}
373
374impl LocalBufferStore {
375 fn save_local_buffer(
376 &self,
377 buffer_handle: Entity<Buffer>,
378 worktree: Entity<Worktree>,
379 path: Arc<RelPath>,
380 mut has_changed_file: bool,
381 cx: &mut Context<BufferStore>,
382 ) -> Task<Result<()>> {
383 let buffer = buffer_handle.read(cx);
384
385 let text = buffer.as_rope().clone();
386 let line_ending = buffer.line_ending();
387 let version = buffer.version();
388 let buffer_id = buffer.remote_id();
389 let file = buffer.file().cloned();
390 let encoding = buffer.encoding.clone();
391
392 if file
393 .as_ref()
394 .is_some_and(|file| file.disk_state() == DiskState::New)
395 {
396 has_changed_file = true;
397 }
398
399 let save = worktree.update(cx, |worktree, cx| {
400 worktree.write_file(path.clone(), text, line_ending, cx, (*encoding).clone())
401 });
402
403 cx.spawn(async move |this, cx| {
404 let new_file = save.await?;
405 let mtime = new_file.disk_state().mtime();
406 this.update(cx, |this, cx| {
407 if let Some((downstream_client, project_id)) = this.downstream_client.clone() {
408 if has_changed_file {
409 downstream_client
410 .send(proto::UpdateBufferFile {
411 project_id,
412 buffer_id: buffer_id.to_proto(),
413 file: Some(language::File::to_proto(&*new_file, cx)),
414 })
415 .log_err();
416 }
417 downstream_client
418 .send(proto::BufferSaved {
419 project_id,
420 buffer_id: buffer_id.to_proto(),
421 version: serialize_version(&version),
422 mtime: mtime.map(|time| time.into()),
423 })
424 .log_err();
425 }
426 })?;
427 buffer_handle.update(cx, |buffer, cx| {
428 if has_changed_file {
429 buffer.file_updated(new_file, cx);
430 }
431 buffer.did_save(version.clone(), mtime, cx);
432 })
433 })
434 }
435
436 fn subscribe_to_worktree(
437 &mut self,
438 worktree: &Entity<Worktree>,
439 cx: &mut Context<BufferStore>,
440 ) {
441 cx.subscribe(worktree, |this, worktree, event, cx| {
442 if worktree.read(cx).is_local()
443 && let worktree::Event::UpdatedEntries(changes) = event
444 {
445 Self::local_worktree_entries_changed(this, &worktree, changes, cx);
446 }
447 })
448 .detach();
449 }
450
451 fn local_worktree_entries_changed(
452 this: &mut BufferStore,
453 worktree_handle: &Entity<Worktree>,
454 changes: &[(Arc<RelPath>, ProjectEntryId, PathChange)],
455 cx: &mut Context<BufferStore>,
456 ) {
457 let snapshot = worktree_handle.read(cx).snapshot();
458 for (path, entry_id, _) in changes {
459 Self::local_worktree_entry_changed(
460 this,
461 *entry_id,
462 path,
463 worktree_handle,
464 &snapshot,
465 cx,
466 );
467 }
468 }
469
470 fn local_worktree_entry_changed(
471 this: &mut BufferStore,
472 entry_id: ProjectEntryId,
473 path: &Arc<RelPath>,
474 worktree: &Entity<worktree::Worktree>,
475 snapshot: &worktree::Snapshot,
476 cx: &mut Context<BufferStore>,
477 ) -> Option<()> {
478 let project_path = ProjectPath {
479 worktree_id: snapshot.id(),
480 path: path.clone(),
481 };
482
483 let buffer_id = this
484 .as_local_mut()
485 .and_then(|local| local.local_buffer_ids_by_entry_id.get(&entry_id))
486 .copied()
487 .or_else(|| this.path_to_buffer_id.get(&project_path).copied())?;
488
489 let buffer = if let Some(buffer) = this.get(buffer_id) {
490 Some(buffer)
491 } else {
492 this.opened_buffers.remove(&buffer_id);
493 this.non_searchable_buffers.remove(&buffer_id);
494 None
495 };
496
497 let buffer = if let Some(buffer) = buffer {
498 buffer
499 } else {
500 this.path_to_buffer_id.remove(&project_path);
501 let this = this.as_local_mut()?;
502 this.local_buffer_ids_by_entry_id.remove(&entry_id);
503 return None;
504 };
505
506 let events = buffer.update(cx, |buffer, cx| {
507 let file = buffer.file()?;
508 let old_file = File::from_dyn(Some(file))?;
509 if old_file.worktree != *worktree {
510 return None;
511 }
512
513 let snapshot_entry = old_file
514 .entry_id
515 .and_then(|entry_id| snapshot.entry_for_id(entry_id))
516 .or_else(|| snapshot.entry_for_path(old_file.path.as_ref()));
517
518 let new_file = if let Some(entry) = snapshot_entry {
519 File {
520 disk_state: match entry.mtime {
521 Some(mtime) => DiskState::Present { mtime },
522 None => old_file.disk_state,
523 },
524 is_local: true,
525 entry_id: Some(entry.id),
526 path: entry.path.clone(),
527 worktree: worktree.clone(),
528 is_private: entry.is_private,
529 encoding: None,
530 }
531 } else {
532 File {
533 disk_state: DiskState::Deleted,
534 is_local: true,
535 entry_id: old_file.entry_id,
536 path: old_file.path.clone(),
537 worktree: worktree.clone(),
538 is_private: old_file.is_private,
539 encoding: None,
540 }
541 };
542
543 if new_file == *old_file {
544 return None;
545 }
546
547 let mut events = Vec::new();
548 if new_file.path != old_file.path {
549 this.path_to_buffer_id.remove(&ProjectPath {
550 path: old_file.path.clone(),
551 worktree_id: old_file.worktree_id(cx),
552 });
553 this.path_to_buffer_id.insert(
554 ProjectPath {
555 worktree_id: new_file.worktree_id(cx),
556 path: new_file.path.clone(),
557 },
558 buffer_id,
559 );
560 events.push(BufferStoreEvent::BufferChangedFilePath {
561 buffer: cx.entity(),
562 old_file: buffer.file().cloned(),
563 });
564 }
565 let local = this.as_local_mut()?;
566 if new_file.entry_id != old_file.entry_id {
567 if let Some(entry_id) = old_file.entry_id {
568 local.local_buffer_ids_by_entry_id.remove(&entry_id);
569 }
570 if let Some(entry_id) = new_file.entry_id {
571 local
572 .local_buffer_ids_by_entry_id
573 .insert(entry_id, buffer_id);
574 }
575 }
576
577 if let Some((client, project_id)) = &this.downstream_client {
578 client
579 .send(proto::UpdateBufferFile {
580 project_id: *project_id,
581 buffer_id: buffer_id.to_proto(),
582 file: Some(new_file.to_proto(cx)),
583 })
584 .ok();
585 }
586
587 buffer.file_updated(Arc::new(new_file), cx);
588 Some(events)
589 })?;
590
591 for event in events {
592 cx.emit(event);
593 }
594
595 None
596 }
597
598 fn save_buffer(
599 &self,
600 buffer: Entity<Buffer>,
601 cx: &mut Context<BufferStore>,
602 ) -> Task<Result<()>> {
603 let Some(file) = File::from_dyn(buffer.read(cx).file()) else {
604 return Task::ready(Err(anyhow!("buffer doesn't have a file")));
605 };
606 let worktree = file.worktree.clone();
607 self.save_local_buffer(buffer, worktree, file.path.clone(), false, cx)
608 }
609
610 fn save_buffer_as(
611 &self,
612 buffer: Entity<Buffer>,
613 path: ProjectPath,
614 cx: &mut Context<BufferStore>,
615 ) -> Task<Result<()>> {
616 let Some(worktree) = self
617 .worktree_store
618 .read(cx)
619 .worktree_for_id(path.worktree_id, cx)
620 else {
621 return Task::ready(Err(anyhow!("no such worktree")));
622 };
623 self.save_local_buffer(buffer, worktree, path.path, true, cx)
624 }
625
626 fn open_buffer(
627 &self,
628 path: Arc<RelPath>,
629 worktree: Entity<Worktree>,
630 encoding: Option<Encoding>,
631 force: bool,
632 detect_utf16: bool,
633 cx: &mut Context<BufferStore>,
634 ) -> Task<Result<Entity<Buffer>>> {
635 let load_buffer = worktree.update(cx, |worktree, cx| {
636 let reservation = cx.reserve_entity();
637 let buffer_id = BufferId::from(reservation.entity_id().as_non_zero_u64());
638
639 let load_file_task = worktree.load_file(
640 path.as_ref(),
641 encoding.clone(),
642 force,
643 detect_utf16,
644 None,
645 cx,
646 );
647
648 cx.spawn(async move |_, cx| {
649 let loaded_file = load_file_task.await?;
650
651 let buffer = cx.insert_entity(reservation, |_| {
652 Buffer::build(
653 text::Buffer::new(ReplicaId::LOCAL, buffer_id, loaded_file.text),
654 Some(loaded_file.file),
655 Capability::ReadWrite,
656 )
657 })?;
658
659 Ok(buffer)
660 })
661 });
662
663 cx.spawn(async move |this, cx| {
664 let buffer = match load_buffer.await {
665 Ok(buffer) => Ok(buffer),
666 Err(error) if is_not_found_error(&error) => cx.new(|cx| {
667 let buffer_id = BufferId::from(cx.entity_id().as_non_zero_u64());
668 let text_buffer = text::Buffer::new(
669 ReplicaId::LOCAL,
670 buffer_id,
671 "",
672 cx.background_executor(),
673 );
674 Buffer::build(
675 text_buffer,
676 Some(Arc::new(File {
677 worktree,
678 path,
679 disk_state: DiskState::New,
680 entry_id: None,
681 is_local: true,
682 is_private: false,
683 encoding: Some(Arc::new(if let Some(encoding) = &encoding {
684 encoding.clone()
685 } else {
686 Encoding::default()
687 })),
688 })),
689 Capability::ReadWrite,
690 )
691 })?,
692 Err(e) => return Err(e),
693 };
694 this.update(cx, |this, cx| {
695 this.add_buffer(buffer.clone(), cx)?;
696 let buffer_id = buffer.read(cx).remote_id();
697 if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
698 this.path_to_buffer_id.insert(
699 ProjectPath {
700 worktree_id: file.worktree_id(cx),
701 path: file.path.clone(),
702 },
703 buffer_id,
704 );
705 let this = this.as_local_mut().unwrap();
706 if let Some(entry_id) = file.entry_id {
707 this.local_buffer_ids_by_entry_id
708 .insert(entry_id, buffer_id);
709 }
710 }
711
712 anyhow::Ok(())
713 })??;
714
715 buffer.update(cx, |buffer, _| {
716 buffer
717 .encoding
718 .set(encoding.unwrap_or(Encoding::default()).get())
719 })?;
720
721 Ok(buffer)
722 })
723 }
724
725 fn create_buffer(
726 &self,
727 project_searchable: bool,
728 cx: &mut Context<BufferStore>,
729 ) -> Task<Result<Entity<Buffer>>> {
730 cx.spawn(async move |buffer_store, cx| {
731 let buffer =
732 cx.new(|cx| Buffer::local("", cx).with_language(language::PLAIN_TEXT.clone(), cx))?;
733 buffer_store.update(cx, |buffer_store, cx| {
734 buffer_store.add_buffer(buffer.clone(), cx).log_err();
735 if !project_searchable {
736 buffer_store
737 .non_searchable_buffers
738 .insert(buffer.read(cx).remote_id());
739 }
740 })?;
741 Ok(buffer)
742 })
743 }
744
745 fn reload_buffers(
746 &self,
747 buffers: HashSet<Entity<Buffer>>,
748 push_to_history: bool,
749 cx: &mut Context<BufferStore>,
750 ) -> Task<Result<ProjectTransaction>> {
751 cx.spawn(async move |_, cx| {
752 let mut project_transaction = ProjectTransaction::default();
753 for buffer in buffers {
754 let transaction = buffer.update(cx, |buffer, cx| buffer.reload(cx))?.await?;
755 buffer.update(cx, |buffer, cx| {
756 if let Some(transaction) = transaction {
757 if !push_to_history {
758 buffer.forget_transaction(transaction.id);
759 }
760 project_transaction.0.insert(cx.entity(), transaction);
761 }
762 })?;
763 }
764
765 Ok(project_transaction)
766 })
767 }
768}
769
770impl BufferStore {
771 pub fn init(client: &AnyProtoClient) {
772 client.add_entity_message_handler(Self::handle_buffer_reloaded);
773 client.add_entity_message_handler(Self::handle_buffer_saved);
774 client.add_entity_message_handler(Self::handle_update_buffer_file);
775 client.add_entity_request_handler(Self::handle_save_buffer);
776 client.add_entity_request_handler(Self::handle_reload_buffers);
777 }
778
779 /// Creates a buffer store, optionally retaining its buffers.
780 pub fn local(worktree_store: Entity<WorktreeStore>, cx: &mut Context<Self>) -> Self {
781 Self {
782 state: BufferStoreState::Local(LocalBufferStore {
783 local_buffer_ids_by_entry_id: Default::default(),
784 worktree_store: worktree_store.clone(),
785 _subscription: cx.subscribe(&worktree_store, |this, _, event, cx| {
786 if let WorktreeStoreEvent::WorktreeAdded(worktree) = event {
787 let this = this.as_local_mut().unwrap();
788 this.subscribe_to_worktree(worktree, cx);
789 }
790 }),
791 }),
792 downstream_client: None,
793 opened_buffers: Default::default(),
794 path_to_buffer_id: Default::default(),
795 shared_buffers: Default::default(),
796 loading_buffers: Default::default(),
797 non_searchable_buffers: Default::default(),
798 worktree_store,
799 }
800 }
801
802 pub fn remote(
803 worktree_store: Entity<WorktreeStore>,
804 upstream_client: AnyProtoClient,
805 remote_id: u64,
806 _cx: &mut Context<Self>,
807 ) -> Self {
808 Self {
809 state: BufferStoreState::Remote(RemoteBufferStore {
810 shared_with_me: Default::default(),
811 loading_remote_buffers_by_id: Default::default(),
812 remote_buffer_listeners: Default::default(),
813 project_id: remote_id,
814 upstream_client,
815 worktree_store: worktree_store.clone(),
816 }),
817 downstream_client: None,
818 opened_buffers: Default::default(),
819 path_to_buffer_id: Default::default(),
820 loading_buffers: Default::default(),
821 shared_buffers: Default::default(),
822 non_searchable_buffers: Default::default(),
823 worktree_store,
824 }
825 }
826
827 fn as_local_mut(&mut self) -> Option<&mut LocalBufferStore> {
828 match &mut self.state {
829 BufferStoreState::Local(state) => Some(state),
830 _ => None,
831 }
832 }
833
834 fn as_remote_mut(&mut self) -> Option<&mut RemoteBufferStore> {
835 match &mut self.state {
836 BufferStoreState::Remote(state) => Some(state),
837 _ => None,
838 }
839 }
840
841 fn as_remote(&self) -> Option<&RemoteBufferStore> {
842 match &self.state {
843 BufferStoreState::Remote(state) => Some(state),
844 _ => None,
845 }
846 }
847
848 pub fn open_buffer(
849 &mut self,
850 project_path: ProjectPath,
851 encoding: Option<Encoding>,
852 force: bool,
853 detect_utf16: bool,
854 cx: &mut Context<Self>,
855 ) -> Task<Result<Entity<Buffer>>> {
856 if let Some(buffer) = self.get_by_path(&project_path) {
857 cx.emit(BufferStoreEvent::BufferOpened {
858 buffer: buffer.clone(),
859 project_path,
860 });
861
862 return Task::ready(Ok(buffer));
863 }
864
865 let task = match self.loading_buffers.entry(project_path.clone()) {
866 hash_map::Entry::Occupied(e) => e.get().clone(),
867 hash_map::Entry::Vacant(entry) => {
868 let path = project_path.path.clone();
869 let Some(worktree) = self
870 .worktree_store
871 .read(cx)
872 .worktree_for_id(project_path.worktree_id, cx)
873 else {
874 return Task::ready(Err(anyhow!("no such worktree")));
875 };
876 let load_buffer = match &self.state {
877 BufferStoreState::Local(this) => {
878 this.open_buffer(path, worktree, encoding, force, detect_utf16, cx)
879 }
880 BufferStoreState::Remote(this) => this.open_buffer(path, worktree, cx),
881 };
882
883 entry
884 .insert(
885 // todo(lw): hot foreground spawn
886 cx.spawn(async move |this, cx| {
887 let load_result = load_buffer.await;
888 this.update(cx, |this, cx| {
889 // Record the fact that the buffer is no longer loading.
890 this.loading_buffers.remove(&project_path);
891
892 let buffer = load_result.map_err(Arc::new)?;
893 cx.emit(BufferStoreEvent::BufferOpened {
894 buffer: buffer.clone(),
895 project_path,
896 });
897
898 Ok(buffer)
899 })?
900 })
901 .shared(),
902 )
903 .clone()
904 }
905 };
906
907 cx.background_spawn(async move {
908 task.await.map_err(|e| {
909 if e.error_code() != ErrorCode::Internal {
910 anyhow!(e.error_code())
911 } else {
912 anyhow!("{e}")
913 }
914 })
915 })
916 }
917
918 pub fn create_buffer(
919 &mut self,
920 project_searchable: bool,
921 cx: &mut Context<Self>,
922 ) -> Task<Result<Entity<Buffer>>> {
923 match &self.state {
924 BufferStoreState::Local(this) => this.create_buffer(project_searchable, cx),
925 BufferStoreState::Remote(this) => this.create_buffer(project_searchable, cx),
926 }
927 }
928
929 pub fn save_buffer(
930 &mut self,
931 buffer: Entity<Buffer>,
932 cx: &mut Context<Self>,
933 ) -> Task<Result<()>> {
934 match &mut self.state {
935 BufferStoreState::Local(this) => this.save_buffer(buffer, cx),
936 BufferStoreState::Remote(this) => this.save_remote_buffer(buffer, None, cx),
937 }
938 }
939
940 pub fn save_buffer_as(
941 &mut self,
942 buffer: Entity<Buffer>,
943 path: ProjectPath,
944 cx: &mut Context<Self>,
945 ) -> Task<Result<()>> {
946 let old_file = buffer.read(cx).file().cloned();
947 let task = match &self.state {
948 BufferStoreState::Local(this) => this.save_buffer_as(buffer.clone(), path, cx),
949 BufferStoreState::Remote(this) => {
950 this.save_remote_buffer(buffer.clone(), Some(path.to_proto()), cx)
951 }
952 };
953 cx.spawn(async move |this, cx| {
954 task.await?;
955 this.update(cx, |this, cx| {
956 old_file.clone().and_then(|file| {
957 this.path_to_buffer_id.remove(&ProjectPath {
958 worktree_id: file.worktree_id(cx),
959 path: file.path().clone(),
960 })
961 });
962
963 cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });
964 })
965 })
966 }
967
968 fn add_buffer(&mut self, buffer_entity: Entity<Buffer>, cx: &mut Context<Self>) -> Result<()> {
969 let buffer = buffer_entity.read(cx);
970 let remote_id = buffer.remote_id();
971 let path = File::from_dyn(buffer.file()).map(|file| ProjectPath {
972 path: file.path.clone(),
973 worktree_id: file.worktree_id(cx),
974 });
975 let is_remote = buffer.replica_id().is_remote();
976 let open_buffer = OpenBuffer::Complete {
977 buffer: buffer_entity.downgrade(),
978 };
979
980 let handle = cx.entity().downgrade();
981 buffer_entity.update(cx, move |_, cx| {
982 cx.on_release(move |buffer, cx| {
983 handle
984 .update(cx, |_, cx| {
985 cx.emit(BufferStoreEvent::BufferDropped(buffer.remote_id()))
986 })
987 .ok();
988 })
989 .detach()
990 });
991 let _expect_path_to_exist;
992 match self.opened_buffers.entry(remote_id) {
993 hash_map::Entry::Vacant(entry) => {
994 entry.insert(open_buffer);
995 _expect_path_to_exist = false;
996 }
997 hash_map::Entry::Occupied(mut entry) => {
998 if let OpenBuffer::Operations(operations) = entry.get_mut() {
999 buffer_entity.update(cx, |b, cx| b.apply_ops(operations.drain(..), cx));
1000 } else if entry.get().upgrade().is_some() {
1001 if is_remote {
1002 return Ok(());
1003 } else {
1004 debug_panic!("buffer {remote_id} was already registered");
1005 anyhow::bail!("buffer {remote_id} was already registered");
1006 }
1007 }
1008 entry.insert(open_buffer);
1009 _expect_path_to_exist = true;
1010 }
1011 }
1012
1013 if let Some(path) = path {
1014 self.path_to_buffer_id.insert(path, remote_id);
1015 }
1016
1017 cx.subscribe(&buffer_entity, Self::on_buffer_event).detach();
1018 cx.emit(BufferStoreEvent::BufferAdded(buffer_entity));
1019 Ok(())
1020 }
1021
1022 pub fn buffers(&self) -> impl '_ + Iterator<Item = Entity<Buffer>> {
1023 self.opened_buffers
1024 .values()
1025 .filter_map(|buffer| buffer.upgrade())
1026 }
1027
1028 pub fn loading_buffers(
1029 &self,
1030 ) -> impl Iterator<Item = (&ProjectPath, impl Future<Output = Result<Entity<Buffer>>>)> {
1031 self.loading_buffers.iter().map(|(path, task)| {
1032 let task = task.clone();
1033 (path, async move {
1034 task.await.map_err(|e| {
1035 if e.error_code() != ErrorCode::Internal {
1036 anyhow!(e.error_code())
1037 } else {
1038 anyhow!("{e}")
1039 }
1040 })
1041 })
1042 })
1043 }
1044
1045 pub fn buffer_id_for_project_path(&self, project_path: &ProjectPath) -> Option<&BufferId> {
1046 self.path_to_buffer_id.get(project_path)
1047 }
1048
1049 pub fn get_by_path(&self, path: &ProjectPath) -> Option<Entity<Buffer>> {
1050 self.path_to_buffer_id
1051 .get(path)
1052 .and_then(|buffer_id| self.get(*buffer_id))
1053 }
1054
1055 pub fn get(&self, buffer_id: BufferId) -> Option<Entity<Buffer>> {
1056 self.opened_buffers.get(&buffer_id)?.upgrade()
1057 }
1058
1059 pub fn get_existing(&self, buffer_id: BufferId) -> Result<Entity<Buffer>> {
1060 self.get(buffer_id)
1061 .with_context(|| format!("unknown buffer id {buffer_id}"))
1062 }
1063
1064 pub fn get_possibly_incomplete(&self, buffer_id: BufferId) -> Option<Entity<Buffer>> {
1065 self.get(buffer_id).or_else(|| {
1066 self.as_remote()
1067 .and_then(|remote| remote.loading_remote_buffers_by_id.get(&buffer_id).cloned())
1068 })
1069 }
1070
1071 pub fn buffer_version_info(&self, cx: &App) -> (Vec<proto::BufferVersion>, Vec<BufferId>) {
1072 let buffers = self
1073 .buffers()
1074 .map(|buffer| {
1075 let buffer = buffer.read(cx);
1076 proto::BufferVersion {
1077 id: buffer.remote_id().into(),
1078 version: language::proto::serialize_version(&buffer.version),
1079 }
1080 })
1081 .collect();
1082 let incomplete_buffer_ids = self
1083 .as_remote()
1084 .map(|remote| remote.incomplete_buffer_ids())
1085 .unwrap_or_default();
1086 (buffers, incomplete_buffer_ids)
1087 }
1088
1089 pub fn disconnected_from_host(&mut self, cx: &mut App) {
1090 for open_buffer in self.opened_buffers.values_mut() {
1091 if let Some(buffer) = open_buffer.upgrade() {
1092 buffer.update(cx, |buffer, _| buffer.give_up_waiting());
1093 }
1094 }
1095
1096 for buffer in self.buffers() {
1097 buffer.update(cx, |buffer, cx| {
1098 buffer.set_capability(Capability::ReadOnly, cx)
1099 });
1100 }
1101
1102 if let Some(remote) = self.as_remote_mut() {
1103 // Wake up all futures currently waiting on a buffer to get opened,
1104 // to give them a chance to fail now that we've disconnected.
1105 remote.remote_buffer_listeners.clear()
1106 }
1107 }
1108
1109 pub fn shared(&mut self, remote_id: u64, downstream_client: AnyProtoClient, _cx: &mut App) {
1110 self.downstream_client = Some((downstream_client, remote_id));
1111 }
1112
1113 pub fn unshared(&mut self, _cx: &mut Context<Self>) {
1114 self.downstream_client.take();
1115 self.forget_shared_buffers();
1116 }
1117
1118 pub fn discard_incomplete(&mut self) {
1119 self.opened_buffers
1120 .retain(|_, buffer| !matches!(buffer, OpenBuffer::Operations(_)));
1121 }
1122
1123 fn buffer_changed_file(&mut self, buffer: Entity<Buffer>, cx: &mut App) -> Option<()> {
1124 let file = File::from_dyn(buffer.read(cx).file())?;
1125
1126 let remote_id = buffer.read(cx).remote_id();
1127 if let Some(entry_id) = file.entry_id {
1128 if let Some(local) = self.as_local_mut() {
1129 match local.local_buffer_ids_by_entry_id.get(&entry_id) {
1130 Some(_) => {
1131 return None;
1132 }
1133 None => {
1134 local
1135 .local_buffer_ids_by_entry_id
1136 .insert(entry_id, remote_id);
1137 }
1138 }
1139 }
1140 self.path_to_buffer_id.insert(
1141 ProjectPath {
1142 worktree_id: file.worktree_id(cx),
1143 path: file.path.clone(),
1144 },
1145 remote_id,
1146 );
1147 };
1148
1149 Some(())
1150 }
1151
1152 pub fn find_search_candidates(
1153 &mut self,
1154 query: &SearchQuery,
1155 mut limit: usize,
1156 fs: Arc<dyn Fs>,
1157 cx: &mut Context<Self>,
1158 ) -> Receiver<Entity<Buffer>> {
1159 let (tx, rx) = smol::channel::unbounded();
1160 let mut open_buffers = HashSet::default();
1161 let mut unnamed_buffers = Vec::new();
1162 for handle in self.buffers() {
1163 let buffer = handle.read(cx);
1164 if self.non_searchable_buffers.contains(&buffer.remote_id()) {
1165 continue;
1166 } else if let Some(entry_id) = buffer.entry_id(cx) {
1167 open_buffers.insert(entry_id);
1168 } else {
1169 limit = limit.saturating_sub(1);
1170 unnamed_buffers.push(handle)
1171 };
1172 }
1173
1174 const MAX_CONCURRENT_BUFFER_OPENS: usize = 64;
1175 let project_paths_rx = self
1176 .worktree_store
1177 .update(cx, |worktree_store, cx| {
1178 worktree_store.find_search_candidates(query.clone(), limit, open_buffers, fs, cx)
1179 })
1180 .chunks(MAX_CONCURRENT_BUFFER_OPENS);
1181
1182 cx.spawn(async move |this, cx| {
1183 for buffer in unnamed_buffers {
1184 tx.send(buffer).await.ok();
1185 }
1186
1187 let mut project_paths_rx = pin!(project_paths_rx);
1188 while let Some(project_paths) = project_paths_rx.next().await {
1189 let buffers = this.update(cx, |this, cx| {
1190 project_paths
1191 .into_iter()
1192 .map(|project_path| this.open_buffer(project_path, None, cx))
1193 .collect::<Vec<_>>()
1194 })?;
1195 for buffer_task in buffers {
1196 if let Some(buffer) = buffer_task.await.log_err()
1197 && tx.send(buffer).await.is_err()
1198 {
1199 return anyhow::Ok(());
1200 }
1201 }
1202 }
1203 anyhow::Ok(())
1204 })
1205 .detach();
1206 rx
1207 }
1208
1209 fn on_buffer_event(
1210 &mut self,
1211 buffer: Entity<Buffer>,
1212 event: &BufferEvent,
1213 cx: &mut Context<Self>,
1214 ) {
1215 match event {
1216 BufferEvent::FileHandleChanged => {
1217 self.buffer_changed_file(buffer, cx);
1218 }
1219 BufferEvent::Reloaded => {
1220 let Some((downstream_client, project_id)) = self.downstream_client.as_ref() else {
1221 return;
1222 };
1223 let buffer = buffer.read(cx);
1224 downstream_client
1225 .send(proto::BufferReloaded {
1226 project_id: *project_id,
1227 buffer_id: buffer.remote_id().to_proto(),
1228 version: serialize_version(&buffer.version()),
1229 mtime: buffer.saved_mtime().map(|t| t.into()),
1230 line_ending: serialize_line_ending(buffer.line_ending()) as i32,
1231 })
1232 .log_err();
1233 }
1234 BufferEvent::LanguageChanged => {}
1235 _ => {}
1236 }
1237 }
1238
1239 pub async fn handle_update_buffer(
1240 this: Entity<Self>,
1241 envelope: TypedEnvelope<proto::UpdateBuffer>,
1242 mut cx: AsyncApp,
1243 ) -> Result<proto::Ack> {
1244 let payload = envelope.payload;
1245 let buffer_id = BufferId::new(payload.buffer_id)?;
1246 let ops = payload
1247 .operations
1248 .into_iter()
1249 .map(language::proto::deserialize_operation)
1250 .collect::<Result<Vec<_>, _>>()?;
1251 this.update(&mut cx, |this, cx| {
1252 match this.opened_buffers.entry(buffer_id) {
1253 hash_map::Entry::Occupied(mut e) => match e.get_mut() {
1254 OpenBuffer::Operations(operations) => operations.extend_from_slice(&ops),
1255 OpenBuffer::Complete { buffer, .. } => {
1256 if let Some(buffer) = buffer.upgrade() {
1257 buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx));
1258 }
1259 }
1260 },
1261 hash_map::Entry::Vacant(e) => {
1262 e.insert(OpenBuffer::Operations(ops));
1263 }
1264 }
1265 Ok(proto::Ack {})
1266 })?
1267 }
1268
1269 pub fn register_shared_lsp_handle(
1270 &mut self,
1271 peer_id: proto::PeerId,
1272 buffer_id: BufferId,
1273 handle: OpenLspBufferHandle,
1274 ) {
1275 if let Some(shared_buffers) = self.shared_buffers.get_mut(&peer_id)
1276 && let Some(buffer) = shared_buffers.get_mut(&buffer_id)
1277 {
1278 buffer.lsp_handle = Some(handle);
1279 return;
1280 }
1281 debug_panic!("tried to register shared lsp handle, but buffer was not shared")
1282 }
1283
1284 pub fn handle_synchronize_buffers(
1285 &mut self,
1286 envelope: TypedEnvelope<proto::SynchronizeBuffers>,
1287 cx: &mut Context<Self>,
1288 client: Arc<Client>,
1289 ) -> Result<proto::SynchronizeBuffersResponse> {
1290 let project_id = envelope.payload.project_id;
1291 let mut response = proto::SynchronizeBuffersResponse {
1292 buffers: Default::default(),
1293 };
1294 let Some(guest_id) = envelope.original_sender_id else {
1295 anyhow::bail!("missing original_sender_id on SynchronizeBuffers request");
1296 };
1297
1298 self.shared_buffers.entry(guest_id).or_default().clear();
1299 for buffer in envelope.payload.buffers {
1300 let buffer_id = BufferId::new(buffer.id)?;
1301 let remote_version = language::proto::deserialize_version(&buffer.version);
1302 if let Some(buffer) = self.get(buffer_id) {
1303 self.shared_buffers
1304 .entry(guest_id)
1305 .or_default()
1306 .entry(buffer_id)
1307 .or_insert_with(|| SharedBuffer {
1308 buffer: buffer.clone(),
1309 lsp_handle: None,
1310 });
1311
1312 let buffer = buffer.read(cx);
1313 response.buffers.push(proto::BufferVersion {
1314 id: buffer_id.into(),
1315 version: language::proto::serialize_version(&buffer.version),
1316 });
1317
1318 let operations = buffer.serialize_ops(Some(remote_version), cx);
1319 let client = client.clone();
1320 if let Some(file) = buffer.file() {
1321 client
1322 .send(proto::UpdateBufferFile {
1323 project_id,
1324 buffer_id: buffer_id.into(),
1325 file: Some(file.to_proto(cx)),
1326 })
1327 .log_err();
1328 }
1329
1330 // TODO(max): do something
1331 // client
1332 // .send(proto::UpdateStagedText {
1333 // project_id,
1334 // buffer_id: buffer_id.into(),
1335 // diff_base: buffer.diff_base().map(ToString::to_string),
1336 // })
1337 // .log_err();
1338
1339 client
1340 .send(proto::BufferReloaded {
1341 project_id,
1342 buffer_id: buffer_id.into(),
1343 version: language::proto::serialize_version(buffer.saved_version()),
1344 mtime: buffer.saved_mtime().map(|time| time.into()),
1345 line_ending: language::proto::serialize_line_ending(buffer.line_ending())
1346 as i32,
1347 })
1348 .log_err();
1349
1350 cx.background_spawn(
1351 async move {
1352 let operations = operations.await;
1353 for chunk in split_operations(operations) {
1354 client
1355 .request(proto::UpdateBuffer {
1356 project_id,
1357 buffer_id: buffer_id.into(),
1358 operations: chunk,
1359 })
1360 .await?;
1361 }
1362 anyhow::Ok(())
1363 }
1364 .log_err(),
1365 )
1366 .detach();
1367 }
1368 }
1369 Ok(response)
1370 }
1371
1372 pub fn handle_create_buffer_for_peer(
1373 &mut self,
1374 envelope: TypedEnvelope<proto::CreateBufferForPeer>,
1375 replica_id: ReplicaId,
1376 capability: Capability,
1377 cx: &mut Context<Self>,
1378 ) -> Result<()> {
1379 let remote = self
1380 .as_remote_mut()
1381 .context("buffer store is not a remote")?;
1382
1383 if let Some(buffer) =
1384 remote.handle_create_buffer_for_peer(envelope, replica_id, capability, cx)?
1385 {
1386 self.add_buffer(buffer, cx)?;
1387 }
1388
1389 Ok(())
1390 }
1391
1392 pub async fn handle_update_buffer_file(
1393 this: Entity<Self>,
1394 envelope: TypedEnvelope<proto::UpdateBufferFile>,
1395 mut cx: AsyncApp,
1396 ) -> Result<()> {
1397 let buffer_id = envelope.payload.buffer_id;
1398 let buffer_id = BufferId::new(buffer_id)?;
1399
1400 this.update(&mut cx, |this, cx| {
1401 let payload = envelope.payload.clone();
1402 if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1403 let file = payload.file.context("invalid file")?;
1404 let worktree = this
1405 .worktree_store
1406 .read(cx)
1407 .worktree_for_id(WorktreeId::from_proto(file.worktree_id), cx)
1408 .context("no such worktree")?;
1409 let file = File::from_proto(file, worktree, cx)?;
1410 let old_file = buffer.update(cx, |buffer, cx| {
1411 let old_file = buffer.file().cloned();
1412 let new_path = file.path.clone();
1413
1414 buffer.file_updated(Arc::new(file), cx);
1415 if old_file.as_ref().is_none_or(|old| *old.path() != new_path) {
1416 Some(old_file)
1417 } else {
1418 None
1419 }
1420 });
1421 if let Some(old_file) = old_file {
1422 cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });
1423 }
1424 }
1425 if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1426 downstream_client
1427 .send(proto::UpdateBufferFile {
1428 project_id: *project_id,
1429 buffer_id: buffer_id.into(),
1430 file: envelope.payload.file,
1431 })
1432 .log_err();
1433 }
1434 Ok(())
1435 })?
1436 }
1437
1438 pub async fn handle_save_buffer(
1439 this: Entity<Self>,
1440 envelope: TypedEnvelope<proto::SaveBuffer>,
1441 mut cx: AsyncApp,
1442 ) -> Result<proto::BufferSaved> {
1443 let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1444 let (buffer, project_id) = this.read_with(&cx, |this, _| {
1445 anyhow::Ok((
1446 this.get_existing(buffer_id)?,
1447 this.downstream_client
1448 .as_ref()
1449 .map(|(_, project_id)| *project_id)
1450 .context("project is not shared")?,
1451 ))
1452 })??;
1453 buffer
1454 .update(&mut cx, |buffer, _| {
1455 buffer.wait_for_version(deserialize_version(&envelope.payload.version))
1456 })?
1457 .await?;
1458 let buffer_id = buffer.read_with(&cx, |buffer, _| buffer.remote_id())?;
1459
1460 if let Some(new_path) = envelope.payload.new_path
1461 && let Some(new_path) = ProjectPath::from_proto(new_path)
1462 {
1463 this.update(&mut cx, |this, cx| {
1464 this.save_buffer_as(buffer.clone(), new_path, cx)
1465 })?
1466 .await?;
1467 } else {
1468 this.update(&mut cx, |this, cx| this.save_buffer(buffer.clone(), cx))?
1469 .await?;
1470 }
1471
1472 buffer.read_with(&cx, |buffer, _| proto::BufferSaved {
1473 project_id,
1474 buffer_id: buffer_id.into(),
1475 version: serialize_version(buffer.saved_version()),
1476 mtime: buffer.saved_mtime().map(|time| time.into()),
1477 })
1478 }
1479
1480 pub async fn handle_close_buffer(
1481 this: Entity<Self>,
1482 envelope: TypedEnvelope<proto::CloseBuffer>,
1483 mut cx: AsyncApp,
1484 ) -> Result<()> {
1485 let peer_id = envelope.sender_id;
1486 let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1487 this.update(&mut cx, |this, cx| {
1488 if let Some(shared) = this.shared_buffers.get_mut(&peer_id)
1489 && shared.remove(&buffer_id).is_some()
1490 {
1491 cx.emit(BufferStoreEvent::SharedBufferClosed(peer_id, buffer_id));
1492 if shared.is_empty() {
1493 this.shared_buffers.remove(&peer_id);
1494 }
1495 return;
1496 }
1497 debug_panic!(
1498 "peer_id {} closed buffer_id {} which was either not open or already closed",
1499 peer_id,
1500 buffer_id
1501 )
1502 })
1503 }
1504
1505 pub async fn handle_buffer_saved(
1506 this: Entity<Self>,
1507 envelope: TypedEnvelope<proto::BufferSaved>,
1508 mut cx: AsyncApp,
1509 ) -> Result<()> {
1510 let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1511 let version = deserialize_version(&envelope.payload.version);
1512 let mtime = envelope.payload.mtime.clone().map(|time| time.into());
1513 this.update(&mut cx, move |this, cx| {
1514 if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1515 buffer.update(cx, |buffer, cx| {
1516 buffer.did_save(version, mtime, cx);
1517 });
1518 }
1519
1520 if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1521 downstream_client
1522 .send(proto::BufferSaved {
1523 project_id: *project_id,
1524 buffer_id: buffer_id.into(),
1525 mtime: envelope.payload.mtime,
1526 version: envelope.payload.version,
1527 })
1528 .log_err();
1529 }
1530 })
1531 }
1532
1533 pub async fn handle_buffer_reloaded(
1534 this: Entity<Self>,
1535 envelope: TypedEnvelope<proto::BufferReloaded>,
1536 mut cx: AsyncApp,
1537 ) -> Result<()> {
1538 let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1539 let version = deserialize_version(&envelope.payload.version);
1540 let mtime = envelope.payload.mtime.clone().map(|time| time.into());
1541 let line_ending = deserialize_line_ending(
1542 proto::LineEnding::from_i32(envelope.payload.line_ending)
1543 .context("missing line ending")?,
1544 );
1545 this.update(&mut cx, |this, cx| {
1546 if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1547 buffer.update(cx, |buffer, cx| {
1548 buffer.did_reload(version, line_ending, mtime, cx);
1549 });
1550 }
1551
1552 if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1553 downstream_client
1554 .send(proto::BufferReloaded {
1555 project_id: *project_id,
1556 buffer_id: buffer_id.into(),
1557 mtime: envelope.payload.mtime,
1558 version: envelope.payload.version,
1559 line_ending: envelope.payload.line_ending,
1560 })
1561 .log_err();
1562 }
1563 })
1564 }
1565
1566 pub fn reload_buffers(
1567 &self,
1568 buffers: HashSet<Entity<Buffer>>,
1569 push_to_history: bool,
1570 cx: &mut Context<Self>,
1571 ) -> Task<Result<ProjectTransaction>> {
1572 if buffers.is_empty() {
1573 return Task::ready(Ok(ProjectTransaction::default()));
1574 }
1575 match &self.state {
1576 BufferStoreState::Local(this) => this.reload_buffers(buffers, push_to_history, cx),
1577 BufferStoreState::Remote(this) => this.reload_buffers(buffers, push_to_history, cx),
1578 }
1579 }
1580
1581 async fn handle_reload_buffers(
1582 this: Entity<Self>,
1583 envelope: TypedEnvelope<proto::ReloadBuffers>,
1584 mut cx: AsyncApp,
1585 ) -> Result<proto::ReloadBuffersResponse> {
1586 let sender_id = envelope.original_sender_id().unwrap_or_default();
1587 let reload = this.update(&mut cx, |this, cx| {
1588 let mut buffers = HashSet::default();
1589 for buffer_id in &envelope.payload.buffer_ids {
1590 let buffer_id = BufferId::new(*buffer_id)?;
1591 buffers.insert(this.get_existing(buffer_id)?);
1592 }
1593 anyhow::Ok(this.reload_buffers(buffers, false, cx))
1594 })??;
1595
1596 let project_transaction = reload.await?;
1597 let project_transaction = this.update(&mut cx, |this, cx| {
1598 this.serialize_project_transaction_for_peer(project_transaction, sender_id, cx)
1599 })?;
1600 Ok(proto::ReloadBuffersResponse {
1601 transaction: Some(project_transaction),
1602 })
1603 }
1604
1605 pub fn create_buffer_for_peer(
1606 &mut self,
1607 buffer: &Entity<Buffer>,
1608 peer_id: proto::PeerId,
1609 cx: &mut Context<Self>,
1610 ) -> Task<Result<()>> {
1611 let buffer_id = buffer.read(cx).remote_id();
1612 let shared_buffers = self.shared_buffers.entry(peer_id).or_default();
1613 if shared_buffers.contains_key(&buffer_id) {
1614 return Task::ready(Ok(()));
1615 }
1616 shared_buffers.insert(
1617 buffer_id,
1618 SharedBuffer {
1619 buffer: buffer.clone(),
1620 lsp_handle: None,
1621 },
1622 );
1623
1624 let Some((client, project_id)) = self.downstream_client.clone() else {
1625 return Task::ready(Ok(()));
1626 };
1627
1628 cx.spawn(async move |this, cx| {
1629 let Some(buffer) = this.read_with(cx, |this, _| this.get(buffer_id))? else {
1630 return anyhow::Ok(());
1631 };
1632
1633 let operations = buffer.update(cx, |b, cx| b.serialize_ops(None, cx))?;
1634 let operations = operations.await;
1635 let state = buffer.update(cx, |buffer, cx| buffer.to_proto(cx))?;
1636
1637 let initial_state = proto::CreateBufferForPeer {
1638 project_id,
1639 peer_id: Some(peer_id),
1640 variant: Some(proto::create_buffer_for_peer::Variant::State(state)),
1641 };
1642
1643 if client.send(initial_state).log_err().is_some() {
1644 let client = client.clone();
1645 cx.background_spawn(async move {
1646 let mut chunks = split_operations(operations).peekable();
1647 while let Some(chunk) = chunks.next() {
1648 let is_last = chunks.peek().is_none();
1649 client.send(proto::CreateBufferForPeer {
1650 project_id,
1651 peer_id: Some(peer_id),
1652 variant: Some(proto::create_buffer_for_peer::Variant::Chunk(
1653 proto::BufferChunk {
1654 buffer_id: buffer_id.into(),
1655 operations: chunk,
1656 is_last,
1657 },
1658 )),
1659 })?;
1660 }
1661 anyhow::Ok(())
1662 })
1663 .await
1664 .log_err();
1665 }
1666 Ok(())
1667 })
1668 }
1669
1670 pub fn forget_shared_buffers(&mut self) {
1671 self.shared_buffers.clear();
1672 }
1673
1674 pub fn forget_shared_buffers_for(&mut self, peer_id: &proto::PeerId) {
1675 self.shared_buffers.remove(peer_id);
1676 }
1677
1678 pub fn update_peer_id(&mut self, old_peer_id: &proto::PeerId, new_peer_id: proto::PeerId) {
1679 if let Some(buffers) = self.shared_buffers.remove(old_peer_id) {
1680 self.shared_buffers.insert(new_peer_id, buffers);
1681 }
1682 }
1683
1684 pub fn has_shared_buffers(&self) -> bool {
1685 !self.shared_buffers.is_empty()
1686 }
1687
1688 pub fn create_local_buffer(
1689 &mut self,
1690 text: &str,
1691 language: Option<Arc<Language>>,
1692 project_searchable: bool,
1693 cx: &mut Context<Self>,
1694 ) -> Entity<Buffer> {
1695 let buffer = cx.new(|cx| {
1696 Buffer::local(text, cx)
1697 .with_language(language.unwrap_or_else(|| language::PLAIN_TEXT.clone()), cx)
1698 });
1699
1700 self.add_buffer(buffer.clone(), cx).log_err();
1701 let buffer_id = buffer.read(cx).remote_id();
1702 if !project_searchable {
1703 self.non_searchable_buffers.insert(buffer_id);
1704 }
1705
1706 if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
1707 self.path_to_buffer_id.insert(
1708 ProjectPath {
1709 worktree_id: file.worktree_id(cx),
1710 path: file.path.clone(),
1711 },
1712 buffer_id,
1713 );
1714 let this = self
1715 .as_local_mut()
1716 .expect("local-only method called in a non-local context");
1717 if let Some(entry_id) = file.entry_id {
1718 this.local_buffer_ids_by_entry_id
1719 .insert(entry_id, buffer_id);
1720 }
1721 }
1722 buffer
1723 }
1724
1725 pub fn deserialize_project_transaction(
1726 &mut self,
1727 message: proto::ProjectTransaction,
1728 push_to_history: bool,
1729 cx: &mut Context<Self>,
1730 ) -> Task<Result<ProjectTransaction>> {
1731 if let Some(this) = self.as_remote_mut() {
1732 this.deserialize_project_transaction(message, push_to_history, cx)
1733 } else {
1734 debug_panic!("not a remote buffer store");
1735 Task::ready(Err(anyhow!("not a remote buffer store")))
1736 }
1737 }
1738
1739 pub fn wait_for_remote_buffer(
1740 &mut self,
1741 id: BufferId,
1742 cx: &mut Context<BufferStore>,
1743 ) -> Task<Result<Entity<Buffer>>> {
1744 if let Some(this) = self.as_remote_mut() {
1745 this.wait_for_remote_buffer(id, cx)
1746 } else {
1747 debug_panic!("not a remote buffer store");
1748 Task::ready(Err(anyhow!("not a remote buffer store")))
1749 }
1750 }
1751
1752 pub fn serialize_project_transaction_for_peer(
1753 &mut self,
1754 project_transaction: ProjectTransaction,
1755 peer_id: proto::PeerId,
1756 cx: &mut Context<Self>,
1757 ) -> proto::ProjectTransaction {
1758 let mut serialized_transaction = proto::ProjectTransaction {
1759 buffer_ids: Default::default(),
1760 transactions: Default::default(),
1761 };
1762 for (buffer, transaction) in project_transaction.0 {
1763 self.create_buffer_for_peer(&buffer, peer_id, cx)
1764 .detach_and_log_err(cx);
1765 serialized_transaction
1766 .buffer_ids
1767 .push(buffer.read(cx).remote_id().into());
1768 serialized_transaction
1769 .transactions
1770 .push(language::proto::serialize_transaction(&transaction));
1771 }
1772 serialized_transaction
1773 }
1774}
1775
1776impl OpenBuffer {
1777 fn upgrade(&self) -> Option<Entity<Buffer>> {
1778 match self {
1779 OpenBuffer::Complete { buffer, .. } => buffer.upgrade(),
1780 OpenBuffer::Operations(_) => None,
1781 }
1782 }
1783}
1784
1785fn is_not_found_error(error: &anyhow::Error) -> bool {
1786 error
1787 .root_cause()
1788 .downcast_ref::<io::Error>()
1789 .is_some_and(|err| err.kind() == io::ErrorKind::NotFound)
1790}