oak-team/oak-gpui
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #360 from zed-industries/improve-files

Browse Source 
Improve remote file handling
This commit is contained in:
Antonio Scandurra
2022-01-24 10:27:01 +01:00
committed by GitHub
parent 2a6f06f18a f859d444ff
commit 5403ab3fd2
8 changed files with 562 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/editor/src/items.rs
+3 -7
View File
@@ -122,13 +122,9 @@ impl ItemView for Editor {
}
fn title(&self, cx: &AppContext) -> String {
let filename = self
.buffer()
.read(cx)
.file(cx)
.and_then(|file| file.file_name());
if let Some(name) = filename {
name.to_string_lossy().into()
let file = self.buffer().read(cx).file(cx);
if let Some(file) = file {
file.file_name(cx).to_string_lossy().into()
} else {
"untitled".into()
}
crates/language/src/buffer.rs
+62 -28
View File
@@ -156,21 +156,20 @@ pub enum Event {
}
pub trait File {
fn as_local(&self) -> Option<&dyn LocalFile>;
fn mtime(&self) -> SystemTime;
/// Returns the path of this file relative to the worktree's root directory.
fn path(&self) -> &Arc<Path>;
/// Returns the absolute path of this file.
fn abs_path(&self) -> Option<PathBuf>;
/// Returns the path of this file relative to the worktree's parent directory (this means it
/// includes the name of the worktree's root folder).
fn full_path(&self) -> PathBuf;
fn full_path(&self, cx: &AppContext) -> PathBuf;
/// Returns the last component of this handle's absolute path. If this handle refers to the root
/// of its worktree, then this method will return the name of the worktree itself.
fn file_name(&self) -> Option<OsString>;
fn file_name(&self, cx: &AppContext) -> OsString;
fn is_deleted(&self) -> bool;
@@ -182,8 +181,6 @@ pub trait File {
cx: &mut MutableAppContext,
) -> Task<Result<(clock::Global, SystemTime)>>;
fn load_local(&self, cx: &AppContext) -> Option<Task<Result<String>>>;
fn format_remote(&self, buffer_id: u64, cx: &mut MutableAppContext)
-> Option<Task<Result<()>>>;
@@ -192,6 +189,23 @@ pub trait File {
fn buffer_removed(&self, buffer_id: u64, cx: &mut MutableAppContext);
fn as_any(&self) -> &dyn Any;
fn to_proto(&self) -> rpc::proto::File;
}
pub trait LocalFile: File {
/// Returns the absolute path of this file.
fn abs_path(&self, cx: &AppContext) -> PathBuf;
fn load(&self, cx: &AppContext) -> Task<Result<String>>;
fn buffer_reloaded(
&self,
buffer_id: u64,
version: &clock::Global,
mtime: SystemTime,
cx: &mut MutableAppContext,
);
}
pub(crate) struct QueryCursorHandle(Option<QueryCursor>);
@@ -348,6 +362,7 @@ impl Buffer {
pub fn to_proto(&self) -> proto::Buffer {
proto::Buffer {
id: self.remote_id(),
file: self.file.as_ref().map(|f| f.to_proto()),
visible_text: self.text.text(),
deleted_text: self.text.deleted_text(),
undo_map: self
@@ -457,7 +472,7 @@ impl Buffer {
if let Some(LanguageServerState { server, .. }) = self.language_server.as_ref() {
let server = server.clone();
let abs_path = file.abs_path().unwrap();
let abs_path = file.as_local().unwrap().abs_path(cx);
let version = self.version();
cx.spawn(|this, mut cx| async move {
let edits = server
@@ -619,7 +634,7 @@ impl Buffer {
None
};
self.update_language_server();
self.update_language_server(cx);
}
pub fn did_save(
@@ -634,7 +649,11 @@ impl Buffer {
if let Some(new_file) = new_file {
self.file = Some(new_file);
}
if let Some(state) = &self.language_server {
if let Some((state, local_file)) = &self
.language_server
.as_ref()
.zip(self.file.as_ref().and_then(|f| f.as_local()))
{
cx.background()
.spawn(
state
@@ -642,10 +661,7 @@ impl Buffer {
.notify::<lsp::notification::DidSaveTextDocument>(
lsp::DidSaveTextDocumentParams {
text_document: lsp::TextDocumentIdentifier {
uri: lsp::Url::from_file_path(
self.file.as_ref().unwrap().abs_path().unwrap(),
)
.unwrap(),
uri: lsp::Url::from_file_path(local_file.abs_path(cx)).unwrap(),
},
text: None,
},
@@ -656,14 +672,33 @@ impl Buffer {
cx.emit(Event::Saved);
}
pub fn did_reload(
&mut self,
version: clock::Global,
mtime: SystemTime,
cx: &mut ModelContext<Self>,
) {
self.saved_mtime = mtime;
self.saved_version = version;
if let Some(file) = self.file.as_ref().and_then(|f| f.as_local()) {
file.buffer_reloaded(self.remote_id(), &self.saved_version, self.saved_mtime, cx);
}
cx.emit(Event::Reloaded);
cx.notify();
}
pub fn file_updated(
&mut self,
new_file: Box<dyn File>,
cx: &mut ModelContext<Self>,
) -> Option<Task<()>> {
let old_file = self.file.as_ref()?;
) -> Task<()> {
let old_file = if let Some(file) = self.file.as_ref() {
file
} else {
return Task::ready(());
};
let mut file_changed = false;
let mut task = None;
let mut task = Task::ready(());
if new_file.path() != old_file.path() {
file_changed = true;
@@ -682,10 +717,12 @@ impl Buffer {
file_changed = true;
if !self.is_dirty() {
task = Some(cx.spawn(|this, mut cx| {
task = cx.spawn(|this, mut cx| {
async move {
let new_text = this.read_with(&cx, |this, cx| {
this.file.as_ref().and_then(|file| file.load_local(cx))
this.file
.as_ref()
.and_then(|file| file.as_local().map(|f| f.load(cx)))
});
if let Some(new_text) = new_text {
let new_text = new_text.await?;
@@ -694,9 +731,7 @@ impl Buffer {
.await;
this.update(&mut cx, |this, cx| {
if this.apply_diff(diff, cx) {
this.saved_version = this.version();
this.saved_mtime = new_mtime;
cx.emit(Event::Reloaded);
this.did_reload(this.version(), new_mtime, cx);
}
});
}
@@ -704,7 +739,7 @@ impl Buffer {
}
.log_err()
.map(drop)
}));
});
}
}
}
@@ -1226,7 +1261,7 @@ impl Buffer {
self.set_active_selections(Arc::from([]), cx);
}
fn update_language_server(&mut self) {
fn update_language_server(&mut self, cx: &AppContext) {
let language_server = if let Some(language_server) = self.language_server.as_mut() {
language_server
} else {
@@ -1235,9 +1270,8 @@ impl Buffer {
let abs_path = self
.file
.as_ref()
.map_or(Path::new("/").to_path_buf(), |file| {
file.abs_path().unwrap()
});
.and_then(|f| f.as_local())
.map_or(Path::new("/").to_path_buf(), |file| file.abs_path(cx));
let version = post_inc(&mut language_server.next_version);
let snapshot = LanguageServerSnapshot {
@@ -1381,7 +1415,7 @@ impl Buffer {
}
self.reparse(cx);
self.update_language_server();
self.update_language_server(cx);
cx.emit(Event::Edited);
if !was_dirty {
crates/project/Cargo.toml
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "project"
version = "0.1.0"
edition = "2018"
edition = "2021"
[lib]
path = "src/project.rs"
crates/project/src/project.rs
+89 -20
View File
@@ -298,6 +298,8 @@ impl Project {
Self::handle_disk_based_diagnostics_updated,
),
client.subscribe_to_entity(remote_id, cx, Self::handle_update_buffer),
client.subscribe_to_entity(remote_id, cx, Self::handle_update_buffer_file),
client.subscribe_to_entity(remote_id, cx, Self::handle_buffer_reloaded),
client.subscribe_to_entity(remote_id, cx, Self::handle_buffer_saved),
],
client,
@@ -624,7 +626,7 @@ impl Project {
) -> Option<()> {
let (path, full_path) = {
let file = buffer.read(cx).file()?;
(file.path().clone(), file.full_path())
(file.path().clone(), file.full_path(cx))
};
// If the buffer has a language, set it and start/assign the language server
@@ -938,7 +940,7 @@ impl Project {
let buffer_abs_path;
if let Some(file) = File::from_dyn(buffer.file()) {
worktree = file.worktree.clone();
buffer_abs_path = file.abs_path();
buffer_abs_path = file.as_local().map(|f| f.abs_path(cx));
} else {
return Task::ready(Err(anyhow!("buffer does not belong to any worktree")));
};
@@ -1136,10 +1138,12 @@ impl Project {
fn add_worktree(&mut self, worktree: &ModelHandle<Worktree>, cx: &mut ModelContext<Self>) {
cx.observe(&worktree, |_, _, cx| cx.notify()).detach();
cx.subscribe(&worktree, |this, worktree, _, cx| {
this.update_open_buffers(worktree, cx)
})
.detach();
if worktree.read(cx).is_local() {
cx.subscribe(&worktree, |this, worktree, _, cx| {
this.update_local_worktree_buffers(worktree, cx);
})
.detach();
}
let push_weak_handle = {
let worktree = worktree.read(cx);
@@ -1161,14 +1165,12 @@ impl Project {
cx.notify();
}
fn update_open_buffers(
fn update_local_worktree_buffers(
&mut self,
worktree_handle: ModelHandle<Worktree>,
cx: &mut ModelContext<Self>,
) {
let local = worktree_handle.read(cx).is_local();
let snapshot = worktree_handle.read(cx).snapshot();
let worktree_path = snapshot.abs_path();
let mut buffers_to_delete = Vec::new();
for (buffer_id, buffer) in &self.open_buffers {
if let OpenBuffer::Loaded(buffer) = buffer {
@@ -1184,8 +1186,7 @@ impl Project {
.and_then(|entry_id| snapshot.entry_for_id(entry_id))
{
File {
is_local: local,
worktree_path: worktree_path.clone(),
is_local: true,
entry_id: Some(entry.id),
mtime: entry.mtime,
path: entry.path.clone(),
@@ -1195,8 +1196,7 @@ impl Project {
snapshot.entry_for_path(old_file.path().as_ref())
{
File {
is_local: local,
worktree_path: worktree_path.clone(),
is_local: true,
entry_id: Some(entry.id),
mtime: entry.mtime,
path: entry.path.clone(),
@@ -1204,8 +1204,7 @@ impl Project {
}
} else {
File {
is_local: local,
worktree_path: worktree_path.clone(),
is_local: true,
entry_id: None,
path: old_file.path().clone(),
mtime: old_file.mtime(),
@@ -1213,9 +1212,18 @@ impl Project {
}
};
if let Some(task) = buffer.file_updated(Box::new(new_file), cx) {
task.detach();
if let Some(project_id) = self.remote_id() {
let client = self.client.clone();
let message = proto::UpdateBufferFile {
project_id,
buffer_id: *buffer_id as u64,
file: Some(new_file.to_proto()),
};
cx.foreground()
.spawn(async move { client.send(message).await })
.detach_and_log_err(cx);
}
buffer.file_updated(Box::new(new_file), cx).detach();
}
});
} else {
@@ -1496,6 +1504,31 @@ impl Project {
Ok(())
}
pub fn handle_update_buffer_file(
&mut self,
envelope: TypedEnvelope<proto::UpdateBufferFile>,
_: Arc<Client>,
cx: &mut ModelContext<Self>,
) -> Result<()> {
let payload = envelope.payload.clone();
let buffer_id = payload.buffer_id as usize;
let file = payload.file.ok_or_else(|| anyhow!("invalid file"))?;
let worktree = self
.worktree_for_id(WorktreeId::from_proto(file.worktree_id), cx)
.ok_or_else(|| anyhow!("no such worktree"))?;
let file = File::from_proto(file, worktree.clone(), cx)?;
let buffer = self
.open_buffers
.get_mut(&buffer_id)
.and_then(|b| b.upgrade(cx))
.ok_or_else(|| anyhow!("no such buffer"))?;
buffer.update(cx, |buffer, cx| {
buffer.file_updated(Box::new(file), cx).detach();
});
Ok(())
}
pub fn handle_save_buffer(
&mut self,
envelope: TypedEnvelope<proto::SaveBuffer>,
@@ -1662,6 +1695,37 @@ impl Project {
Ok(())
}
pub fn handle_buffer_reloaded(
&mut self,
envelope: TypedEnvelope<proto::BufferReloaded>,
_: Arc<Client>,
cx: &mut ModelContext<Self>,
) -> Result<()> {
let payload = envelope.payload.clone();
let buffer = self
.open_buffers
.get(&(payload.buffer_id as usize))
.and_then(|buf| {
if let OpenBuffer::Loaded(buffer) = buf {
buffer.upgrade(cx)
} else {
None
}
});
if let Some(buffer) = buffer {
buffer.update(cx, |buffer, cx| {
let version = payload.version.try_into()?;
let mtime = payload
.mtime
.ok_or_else(|| anyhow!("missing mtime"))?
.into();
buffer.did_reload(version, mtime, cx);
Result::<_, anyhow::Error>::Ok(())
})?;
}
Ok(())
}
pub fn match_paths<'a>(
&self,
query: &'a str,
@@ -2185,8 +2249,13 @@ mod tests {
cx.update(|cx| {
let target_buffer = definition.target_buffer.read(cx);
assert_eq!(
target_buffer.file().unwrap().abs_path(),
Some(dir.path().join("a.rs"))
target_buffer
.file()
.unwrap()
.as_local()
.unwrap()
.abs_path(cx),
dir.path().join("a.rs")
);
assert_eq!(definition.target_range.to_offset(target_buffer), 9..10);
assert_eq!(
@@ -2432,7 +2501,7 @@ mod tests {
.as_remote_mut()
.unwrap()
.snapshot
.apply_update(update_message)
.apply_remote_update(update_message)
.unwrap();
assert_eq!(
crates/project/src/worktree.rs
+192 -128
View File
@@ -30,7 +30,7 @@ use std::{
ffi::{OsStr, OsString},
fmt,
future::Future,
ops::Deref,
ops::{Deref, DerefMut},
path::{Path, PathBuf},
sync::{
atomic::{AtomicUsize, Ordering::SeqCst},
@@ -54,9 +54,9 @@ pub enum Worktree {
}
pub struct LocalWorktree {
snapshot: Snapshot,
snapshot: LocalSnapshot,
config: WorktreeConfig,
background_snapshot: Arc<Mutex<Snapshot>>,
background_snapshot: Arc<Mutex<LocalSnapshot>>,
last_scan_state_rx: watch::Receiver<ScanState>,
_background_scanner_task: Option<Task<()>>,
poll_task: Option<Task<()>>,
@@ -85,15 +85,34 @@ pub struct RemoteWorktree {
#[derive(Clone)]
pub struct Snapshot {
id: WorktreeId,
scan_id: usize,
abs_path: Arc<Path>,
root_name: String,
root_char_bag: CharBag,
ignores: HashMap<Arc<Path>, (Arc<Gitignore>, usize)>,
entries_by_path: SumTree<Entry>,
entries_by_id: SumTree<PathEntry>,
}
#[derive(Clone)]
pub struct LocalSnapshot {
abs_path: Arc<Path>,
scan_id: usize,
ignores: HashMap<Arc<Path>, (Arc<Gitignore>, usize)>,
removed_entry_ids: HashMap<u64, usize>,
next_entry_id: Arc<AtomicUsize>,
snapshot: Snapshot,
}
impl Deref for LocalSnapshot {
type Target = Snapshot;
fn deref(&self) -> &Self::Target {
&self.snapshot
}
}
impl DerefMut for LocalSnapshot {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.snapshot
}
}
#[derive(Clone, Debug)]
@@ -112,7 +131,7 @@ enum Registration {
struct ShareState {
project_id: u64,
snapshots_tx: Sender<Snapshot>,
snapshots_tx: Sender<LocalSnapshot>,
_maintain_remote_snapshot: Option<Task<()>>,
}
@@ -158,7 +177,7 @@ impl Worktree {
let (tree, scan_states_tx) = LocalWorktree::new(client, path, weak, fs.clone(), cx).await?;
tree.update(cx, |tree, cx| {
let tree = tree.as_local_mut().unwrap();
let abs_path = tree.snapshot.abs_path.clone();
let abs_path = tree.abs_path().clone();
let background_snapshot = tree.background_snapshot.clone();
let background = cx.background().clone();
tree._background_scanner_task = Some(cx.background().spawn(async move {
@@ -233,15 +252,10 @@ impl Worktree {
cx.add_model(|cx: &mut ModelContext<Worktree>| {
let snapshot = Snapshot {
id: WorktreeId(remote_id as usize),
scan_id: 0,
abs_path: Path::new("").into(),
root_name,
root_char_bag,
ignores: Default::default(),
entries_by_path,
entries_by_id,
removed_entry_ids: Default::default(),
next_entry_id: Default::default(),
};
let (updates_tx, mut updates_rx) = postage::mpsc::channel(64);
@@ -251,7 +265,7 @@ impl Worktree {
.spawn(async move {
while let Some(update) = updates_rx.recv().await {
let mut snapshot = snapshot_tx.borrow().clone();
if let Err(error) = snapshot.apply_update(update) {
if let Err(error) = snapshot.apply_remote_update(update) {
log::error!("error applying worktree update: {}", error);
}
*snapshot_tx.borrow_mut() = snapshot;
@@ -328,7 +342,7 @@ impl Worktree {
pub fn snapshot(&self) -> Snapshot {
match self {
Worktree::Local(worktree) => worktree.snapshot(),
Worktree::Local(worktree) => worktree.snapshot().snapshot,
Worktree::Remote(worktree) => worktree.snapshot(),
}
}
@@ -469,28 +483,28 @@ impl LocalWorktree {
let (scan_states_tx, scan_states_rx) = smol::channel::unbounded();
let (mut last_scan_state_tx, last_scan_state_rx) = watch::channel_with(ScanState::Scanning);
let tree = cx.add_model(move |cx: &mut ModelContext<Worktree>| {
let mut snapshot = Snapshot {
id: WorktreeId::from_usize(cx.model_id()),
scan_id: 0,
let mut snapshot = LocalSnapshot {
abs_path,
root_name: root_name.clone(),
root_char_bag,
scan_id: 0,
ignores: Default::default(),
entries_by_path: Default::default(),
entries_by_id: Default::default(),
removed_entry_ids: Default::default(),
next_entry_id: Arc::new(next_entry_id),
snapshot: Snapshot {
id: WorktreeId::from_usize(cx.model_id()),
root_name: root_name.clone(),
root_char_bag,
entries_by_path: Default::default(),
entries_by_id: Default::default(),
},
};
if let Some(metadata) = metadata {
snapshot.insert_entry(
Entry::new(
path.into(),
&metadata,
&snapshot.next_entry_id,
snapshot.root_char_bag,
),
fs.as_ref(),
let entry = Entry::new(
path.into(),
&metadata,
&snapshot.next_entry_id,
snapshot.root_char_bag,
);
snapshot.insert_entry(entry, fs.as_ref());
}
let tree = Self {
@@ -543,6 +557,22 @@ impl LocalWorktree {
Ok((tree, scan_states_tx))
}
pub fn abs_path(&self) -> &Arc<Path> {
&self.abs_path
}
pub fn contains_abs_path(&self, path: &Path) -> bool {
path.starts_with(&self.abs_path)
}
fn absolutize(&self, path: &Path) -> PathBuf {
if path.file_name().is_some() {
self.abs_path.join(path)
} else {
self.abs_path.to_path_buf()
}
}
pub fn authorized_logins(&self) -> Vec<String> {
self.config.collaborators.clone()
}
@@ -624,14 +654,13 @@ impl LocalWorktree {
}
}
pub fn snapshot(&self) -> Snapshot {
pub fn snapshot(&self) -> LocalSnapshot {
self.snapshot.clone()
}
fn load(&self, path: &Path, cx: &mut ModelContext<Worktree>) -> Task<Result<(File, String)>> {
let handle = cx.handle();
let path = Arc::from(path);
let worktree_path = self.abs_path.clone();
let abs_path = self.absolutize(&path);
let background_snapshot = self.background_snapshot.clone();
let fs = self.fs.clone();
@@ -644,7 +673,6 @@ impl LocalWorktree {
File {
entry_id: Some(entry.id),
worktree: handle,
worktree_path,
path: entry.path,
mtime: entry.mtime,
is_local: true,
@@ -664,19 +692,16 @@ impl LocalWorktree {
let text = buffer.as_rope().clone();
let version = buffer.version();
let save = self.save(path, text, cx);
cx.spawn(|this, mut cx| async move {
let handle = cx.handle();
cx.as_mut().spawn(|mut cx| async move {
let entry = save.await?;
let file = this.update(&mut cx, |this, cx| {
let this = this.as_local_mut().unwrap();
File {
entry_id: Some(entry.id),
worktree: cx.handle(),
worktree_path: this.abs_path.clone(),
path: entry.path,
mtime: entry.mtime,
is_local: true,
}
});
let file = File {
entry_id: Some(entry.id),
worktree: handle,
path: entry.path,
mtime: entry.mtime,
is_local: true,
};
buffer_handle.update(&mut cx, |buffer, cx| {
buffer.did_save(version, file.mtime, Some(Box::new(file)), cx);
@@ -755,7 +780,8 @@ impl LocalWorktree {
let snapshot = self.snapshot();
let rpc = self.client.clone();
let worktree_id = cx.model_id() as u64;
let (snapshots_to_send_tx, snapshots_to_send_rx) = smol::channel::unbounded::<Snapshot>();
let (snapshots_to_send_tx, snapshots_to_send_rx) =
smol::channel::unbounded::<LocalSnapshot>();
let maintain_remote_snapshot = cx.background().spawn({
let rpc = rpc.clone();
let snapshot = snapshot.clone();
@@ -811,15 +837,9 @@ impl RemoteWorktree {
let replica_id = self.replica_id;
let project_id = self.project_id;
let remote_worktree_id = self.id();
let root_path = self.snapshot.abs_path.clone();
let path: Arc<Path> = Arc::from(path);
let path_string = path.to_string_lossy().to_string();
cx.spawn_weak(move |this, mut cx| async move {
let entry = this
.upgrade(&cx)
.ok_or_else(|| anyhow!("worktree was closed"))?
.read_with(&cx, |tree, _| tree.entry_for_path(&path).cloned())
.ok_or_else(|| anyhow!("file does not exist"))?;
let response = rpc
.request(proto::OpenBuffer {
project_id,
@@ -831,18 +851,15 @@ impl RemoteWorktree {
let this = this
.upgrade(&cx)
.ok_or_else(|| anyhow!("worktree was closed"))?;
let file = File {
entry_id: Some(entry.id),
worktree: this.clone(),
worktree_path: root_path,
path: entry.path,
mtime: entry.mtime,
is_local: false,
};
let remote_buffer = response.buffer.ok_or_else(|| anyhow!("empty buffer"))?;
Ok(cx.add_model(|cx| {
Buffer::from_proto(replica_id, remote_buffer, Some(Box::new(file)), cx).unwrap()
}))
let mut remote_buffer = response.buffer.ok_or_else(|| anyhow!("empty buffer"))?;
let file = remote_buffer
.file
.take()
.map(|proto| cx.read(|cx| File::from_proto(proto, this.clone(), cx)))
.transpose()?
.map(|file| Box::new(file) as Box<dyn language::File>);
Ok(cx.add_model(|cx| Buffer::from_proto(replica_id, remote_buffer, file, cx).unwrap()))
})
}
@@ -976,10 +993,7 @@ impl Snapshot {
}
}
pub(crate) fn apply_update(&mut self, update: proto::UpdateWorktree) -> Result<()> {
self.scan_id += 1;
let scan_id = self.scan_id;
pub(crate) fn apply_remote_update(&mut self, update: proto::UpdateWorktree) -> Result<()> {
let mut entries_by_path_edits = Vec::new();
let mut entries_by_id_edits = Vec::new();
for entry_id in update.removed_entries {
@@ -1000,7 +1014,7 @@ impl Snapshot {
id: entry.id,
path: entry.path.clone(),
is_ignored: entry.is_ignored,
scan_id,
scan_id: 0,
}));
entries_by_path_edits.push(Edit::Insert(entry));
}
@@ -1087,22 +1101,6 @@ impl Snapshot {
}
}
pub fn contains_abs_path(&self, path: &Path) -> bool {
path.starts_with(&self.abs_path)
}
fn absolutize(&self, path: &Path) -> PathBuf {
if path.file_name().is_some() {
self.abs_path.join(path)
} else {
self.abs_path.to_path_buf()
}
}
pub fn abs_path(&self) -> &Arc<Path> {
&self.abs_path
}
pub fn root_entry(&self) -> Option<&Entry> {
self.entry_for_path("")
}
@@ -1132,7 +1130,9 @@ impl Snapshot {
pub fn inode_for_path(&self, path: impl AsRef<Path>) -> Option<u64> {
self.entry_for_path(path.as_ref()).map(|e| e.inode)
}
}
impl LocalSnapshot {
fn insert_entry(&mut self, mut entry: Entry, fs: &dyn Fs) -> Entry {
if !entry.is_dir() && entry.path.file_name() == Some(&GITIGNORE) {
let abs_path = self.abs_path.join(&entry.path);
@@ -1154,12 +1154,13 @@ impl Snapshot {
self.reuse_entry_id(&mut entry);
self.entries_by_path.insert_or_replace(entry.clone(), &());
let scan_id = self.scan_id;
self.entries_by_id.insert_or_replace(
PathEntry {
id: entry.id,
path: entry.path.clone(),
is_ignored: entry.is_ignored,
scan_id: self.scan_id,
scan_id,
},
&(),
);
@@ -1315,7 +1316,7 @@ impl Deref for Worktree {
}
impl Deref for LocalWorktree {
type Target = Snapshot;
type Target = LocalSnapshot;
fn deref(&self) -> &Self::Target {
&self.snapshot
@@ -1354,11 +1355,18 @@ pub struct File {
pub path: Arc<Path>,
pub mtime: SystemTime,
pub(crate) entry_id: Option<usize>,
pub(crate) worktree_path: Arc<Path>,
pub(crate) is_local: bool,
}
impl language::File for File {
fn as_local(&self) -> Option<&dyn language::LocalFile> {
if self.is_local {
Some(self)
} else {
None
}
}
fn mtime(&self) -> SystemTime {
self.mtime
}
@@ -1367,30 +1375,20 @@ impl language::File for File {
&self.path
}
fn abs_path(&self) -> Option<PathBuf> {
if self.is_local {
Some(self.worktree_path.join(&self.path))
} else {
None
}
}
fn full_path(&self) -> PathBuf {
fn full_path(&self, cx: &AppContext) -> PathBuf {
let mut full_path = PathBuf::new();
if let Some(worktree_name) = self.worktree_path.file_name() {
full_path.push(worktree_name);
}
full_path.push(self.worktree.read(cx).root_name());
full_path.push(&self.path);
full_path
}
/// Returns the last component of this handle's absolute path. If this handle refers to the root
/// of its worktree, then this method will return the name of the worktree itself.
fn file_name<'a>(&'a self) -> Option<OsString> {
fn file_name(&self, cx: &AppContext) -> OsString {
self.path
.file_name()
.or_else(|| self.worktree_path.file_name())
.map(Into::into)
.map(|name| name.into())
.unwrap_or_else(|| OsString::from(&self.worktree.read(cx).root_name))
}
fn is_deleted(&self) -> bool {
@@ -1444,16 +1442,6 @@ impl language::File for File {
})
}
fn load_local(&self, cx: &AppContext) -> Option<Task<Result<String>>> {
let worktree = self.worktree.read(cx).as_local()?;
let abs_path = worktree.absolutize(&self.path);
let fs = worktree.fs.clone();
Some(
cx.background()
.spawn(async move { fs.load(&abs_path).await }),
)
}
fn format_remote(
&self,
buffer_id: u64,
@@ -1504,9 +1492,83 @@ impl language::File for File {
fn as_any(&self) -> &dyn Any {
self
}
fn to_proto(&self) -> rpc::proto::File {
rpc::proto::File {
worktree_id: self.worktree.id() as u64,
entry_id: self.entry_id.map(|entry_id| entry_id as u64),
path: self.path.to_string_lossy().into(),
mtime: Some(self.mtime.into()),
}
}
}
impl language::LocalFile for File {
fn abs_path(&self, cx: &AppContext) -> PathBuf {
self.worktree
.read(cx)
.as_local()
.unwrap()
.abs_path
.join(&self.path)
}
fn load(&self, cx: &AppContext) -> Task<Result<String>> {
let worktree = self.worktree.read(cx).as_local().unwrap();
let abs_path = worktree.absolutize(&self.path);
let fs = worktree.fs.clone();
cx.background()
.spawn(async move { fs.load(&abs_path).await })
}
fn buffer_reloaded(
&self,
buffer_id: u64,
version: &clock::Global,
mtime: SystemTime,
cx: &mut MutableAppContext,
) {
let worktree = self.worktree.read(cx).as_local().unwrap();
if let Some(project_id) = worktree.share.as_ref().map(|share| share.project_id) {
let rpc = worktree.client.clone();
let message = proto::BufferReloaded {
project_id,
buffer_id,
version: version.into(),
mtime: Some(mtime.into()),
};
cx.background()
.spawn(async move { rpc.send(message).await })
.detach_and_log_err(cx);
}
}
}
impl File {
pub fn from_proto(
proto: rpc::proto::File,
worktree: ModelHandle<Worktree>,
cx: &AppContext,
) -> Result<Self> {
let worktree_id = worktree
.read(cx)
.as_remote()
.ok_or_else(|| anyhow!("not remote"))?
.id();
if worktree_id.to_proto() != proto.worktree_id {
return Err(anyhow!("worktree id does not match file"));
}
Ok(Self {
worktree,
path: Path::new(&proto.path).into(),
mtime: proto.mtime.ok_or_else(|| anyhow!("no timestamp"))?.into(),
entry_id: proto.entry_id.map(|entry_id| entry_id as usize),
is_local: false,
})
}
pub fn from_dyn(file: Option<&dyn language::File>) -> Option<&Self> {
file.and_then(|f| f.as_any().downcast_ref())
}
@@ -1690,14 +1752,14 @@ impl<'a> sum_tree::Dimension<'a, EntrySummary> for PathKey {
struct BackgroundScanner {
fs: Arc<dyn Fs>,
snapshot: Arc<Mutex<Snapshot>>,
snapshot: Arc<Mutex<LocalSnapshot>>,
notify: Sender<ScanState>,
executor: Arc<executor::Background>,
}
impl BackgroundScanner {
fn new(
snapshot: Arc<Mutex<Snapshot>>,
snapshot: Arc<Mutex<LocalSnapshot>>,
notify: Sender<ScanState>,
fs: Arc<dyn Fs>,
executor: Arc<executor::Background>,
@@ -1714,7 +1776,7 @@ impl BackgroundScanner {
self.snapshot.lock().abs_path.clone()
}
fn snapshot(&self) -> Snapshot {
fn snapshot(&self) -> LocalSnapshot {
self.snapshot.lock().clone()
}
@@ -2057,7 +2119,7 @@ impl BackgroundScanner {
.await;
}
async fn update_ignore_status(&self, job: UpdateIgnoreStatusJob, snapshot: &Snapshot) {
async fn update_ignore_status(&self, job: UpdateIgnoreStatusJob, snapshot: &LocalSnapshot) {
let mut ignore_stack = job.ignore_stack;
if let Some((ignore, _)) = snapshot.ignores.get(&job.path) {
ignore_stack = ignore_stack.append(job.path.clone(), ignore.clone());
@@ -2101,7 +2163,7 @@ impl BackgroundScanner {
async fn refresh_entry(
fs: &dyn Fs,
snapshot: &Mutex<Snapshot>,
snapshot: &Mutex<LocalSnapshot>,
path: Arc<Path>,
abs_path: &Path,
) -> Result<Entry> {
@@ -2169,7 +2231,7 @@ impl WorktreeHandle for ModelHandle<Worktree> {
use smol::future::FutureExt;
let filename = "fs-event-sentinel";
let root_path = cx.read(|cx| self.read(cx).abs_path.clone());
let root_path = cx.read(|cx| self.read(cx).as_local().unwrap().abs_path().clone());
let tree = self.clone();
async move {
std::fs::write(root_path.join(filename), "").unwrap();
@@ -2521,17 +2583,19 @@ mod tests {
let (notify_tx, _notify_rx) = smol::channel::unbounded();
let fs = Arc::new(RealFs);
let next_entry_id = Arc::new(AtomicUsize::new(0));
let mut initial_snapshot = Snapshot {
id: WorktreeId::from_usize(0),
scan_id: 0,
let mut initial_snapshot = LocalSnapshot {
abs_path: root_dir.path().into(),
entries_by_path: Default::default(),
entries_by_id: Default::default(),
scan_id: 0,
removed_entry_ids: Default::default(),
ignores: Default::default(),
root_name: Default::default(),
root_char_bag: Default::default(),
next_entry_id: next_entry_id.clone(),
snapshot: Snapshot {
id: WorktreeId::from_usize(0),
entries_by_path: Default::default(),
entries_by_id: Default::default(),
root_name: Default::default(),
root_char_bag: Default::default(),
},
};
initial_snapshot.insert_entry(
Entry::new(
@@ -2612,7 +2676,7 @@ mod tests {
let update = scanner
.snapshot()
.build_update(&prev_snapshot, 0, 0, include_ignored);
prev_snapshot.apply_update(update).unwrap();
prev_snapshot.apply_remote_update(update).unwrap();
assert_eq!(
prev_snapshot.to_vec(true),
scanner.snapshot().to_vec(include_ignored)
@@ -2767,7 +2831,7 @@ mod tests {
.collect()
}
impl Snapshot {
impl LocalSnapshot {
fn check_invariants(&self) {
let mut files = self.files(true, 0);
let mut visible_files = self.files(false, 0);
crates/rpc/proto/zed.proto
+60 -37
View File
@@ -33,25 +33,27 @@ message Envelope {
OpenBufferResponse open_buffer_response = 25;
CloseBuffer close_buffer = 26;
UpdateBuffer update_buffer = 27;
SaveBuffer save_buffer = 28;
BufferSaved buffer_saved = 29;
FormatBuffer format_buffer = 30;
UpdateBufferFile update_buffer_file = 28;
SaveBuffer save_buffer = 29;
BufferSaved buffer_saved = 30;
BufferReloaded buffer_reloaded = 31;
FormatBuffer format_buffer = 32;
GetChannels get_channels = 31;
GetChannelsResponse get_channels_response = 32;
JoinChannel join_channel = 33;
JoinChannelResponse join_channel_response = 34;
LeaveChannel leave_channel = 35;
SendChannelMessage send_channel_message = 36;
SendChannelMessageResponse send_channel_message_response = 37;
ChannelMessageSent channel_message_sent = 38;
GetChannelMessages get_channel_messages = 39;
GetChannelMessagesResponse get_channel_messages_response = 40;
GetChannels get_channels = 33;
GetChannelsResponse get_channels_response = 34;
JoinChannel join_channel = 35;
JoinChannelResponse join_channel_response = 36;
LeaveChannel leave_channel = 37;
SendChannelMessage send_channel_message = 38;
SendChannelMessageResponse send_channel_message_response = 39;
ChannelMessageSent channel_message_sent = 40;
GetChannelMessages get_channel_messages = 41;
GetChannelMessagesResponse get_channel_messages_response = 42;
UpdateContacts update_contacts = 41;
UpdateContacts update_contacts = 43;
GetUsers get_users = 42;
GetUsersResponse get_users_response = 43;
GetUsers get_users = 44;
GetUsersResponse get_users_response = 45;
}
}
@@ -88,9 +90,9 @@ message JoinProject {
}
message JoinProjectResponse {
uint32 replica_id = 2;
repeated Worktree worktrees = 3;
repeated Collaborator collaborators = 4;
uint32 replica_id = 1;
repeated Worktree worktrees = 2;
repeated Collaborator collaborators = 3;
}
message LeaveProject {
@@ -150,7 +152,13 @@ message CloseBuffer {
message UpdateBuffer {
uint64 project_id = 1;
uint64 buffer_id = 2;
repeated Operation operations = 4;
repeated Operation operations = 3;
}
message UpdateBufferFile {
uint64 project_id = 1;
uint64 buffer_id = 2;
File file = 3;
}
message SaveBuffer {
@@ -165,6 +173,13 @@ message BufferSaved {
Timestamp mtime = 4;
}
message BufferReloaded {
uint64 project_id = 1;
uint64 buffer_id = 2;
repeated VectorClockEntry version = 3;
Timestamp mtime = 4;
}
message FormatBuffer {
uint64 project_id = 1;
uint64 buffer_id = 2;
@@ -177,11 +192,11 @@ message UpdateDiagnosticSummary {
}
message DiagnosticSummary {
string path = 3;
uint32 error_count = 4;
uint32 warning_count = 5;
uint32 info_count = 6;
uint32 hint_count = 7;
string path = 1;
uint32 error_count = 2;
uint32 warning_count = 3;
uint32 info_count = 4;
uint32 hint_count = 5;
}
message DiskBasedDiagnosticsUpdating {
@@ -270,6 +285,13 @@ message Worktree {
bool weak = 5;
}
message File {
uint64 worktree_id = 1;
optional uint64 entry_id = 2;
string path = 3;
Timestamp mtime = 4;
}
message Entry {
uint64 id = 1;
bool is_dir = 2;
@@ -282,15 +304,16 @@ message Entry {
message Buffer {
uint64 id = 1;
string visible_text = 2;
string deleted_text = 3;
repeated BufferFragment fragments = 4;
repeated UndoMapEntry undo_map = 5;
repeated VectorClockEntry version = 6;
repeated SelectionSet selections = 7;
repeated Diagnostic diagnostics = 8;
uint32 lamport_timestamp = 9;
repeated Operation deferred_operations = 10;
optional File file = 2;
string visible_text = 3;
string deleted_text = 4;
repeated BufferFragment fragments = 5;
repeated UndoMapEntry undo_map = 6;
repeated VectorClockEntry version = 7;
repeated SelectionSet selections = 8;
repeated Diagnostic diagnostics = 9;
uint32 lamport_timestamp = 10;
repeated Operation deferred_operations = 11;
}
message BufferFragment {
@@ -299,7 +322,7 @@ message BufferFragment {
uint32 lamport_timestamp = 3;
uint32 insertion_offset = 4;
uint32 len = 5;
bool visible = 6;
bool visible = 6;
repeated VectorClockEntry deletions = 7;
repeated VectorClockEntry max_undos = 8;
}
@@ -383,8 +406,8 @@ message Operation {
message UpdateSelections {
uint32 replica_id = 1;
uint32 lamport_timestamp = 3;
repeated Selection selections = 4;
uint32 lamport_timestamp = 2;
repeated Selection selections = 3;
}
}
crates/rpc/src/proto.rs
+4
View File
@@ -122,6 +122,7 @@ macro_rules! entity_messages {
messages!(
Ack,
AddProjectCollaborator,
BufferReloaded,
BufferSaved,
ChannelMessageSent,
CloseBuffer,
@@ -157,6 +158,7 @@ messages!(
UnregisterWorktree,
UnshareProject,
UpdateBuffer,
UpdateBufferFile,
UpdateContacts,
UpdateDiagnosticSummary,
UpdateWorktree,
@@ -183,6 +185,7 @@ request_messages!(
entity_messages!(
project_id,
AddProjectCollaborator,
BufferReloaded,
BufferSaved,
CloseBuffer,
DiskBasedDiagnosticsUpdated,
@@ -197,6 +200,7 @@ entity_messages!(
UnregisterWorktree,
UnshareProject,
UpdateBuffer,
UpdateBufferFile,
UpdateDiagnosticSummary,
UpdateWorktree,
);
crates/server/src/rpc.rs
+151 -6
View File
@@ -77,6 +77,8 @@ impl Server {
.add_handler(Server::open_buffer)
.add_handler(Server::close_buffer)
.add_handler(Server::update_buffer)
.add_handler(Server::update_buffer_file)
.add_handler(Server::buffer_reloaded)
.add_handler(Server::buffer_saved)
.add_handler(Server::save_buffer)
.add_handler(Server::format_buffer)
@@ -704,6 +706,38 @@ impl Server {
Ok(())
}
async fn update_buffer_file(
self: Arc<Server>,
request: TypedEnvelope<proto::UpdateBufferFile>,
) -> tide::Result<()> {
let receiver_ids = self
.state()
.project_connection_ids(request.payload.project_id, request.sender_id)
.ok_or_else(|| anyhow!(NO_SUCH_PROJECT))?;
broadcast(request.sender_id, receiver_ids, |connection_id| {
self.peer
.forward_send(request.sender_id, connection_id, request.payload.clone())
})
.await?;
Ok(())
}
async fn buffer_reloaded(
self: Arc<Server>,
request: TypedEnvelope<proto::BufferReloaded>,
) -> tide::Result<()> {
let receiver_ids = self
.state()
.project_connection_ids(request.payload.project_id, request.sender_id)
.ok_or_else(|| anyhow!(NO_SUCH_PROJECT))?;
broadcast(request.sender_id, receiver_ids, |connection_id| {
self.peer
.forward_send(request.sender_id, connection_id, request.payload.clone())
})
.await?;
Ok(())
}
async fn buffer_saved(
self: Arc<Server>,
request: TypedEnvelope<proto::BufferSaved>,
@@ -1470,9 +1504,7 @@ mod tests {
.condition(&mut cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
.await;
buffer_b
.condition(&mut cx_b, |buf, _| {
dbg!(buf.text()) == "i-am-c, i-am-b, i-am-a"
})
.condition(&mut cx_b, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
.await;
buffer_c
.condition(&mut cx_c, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
@@ -1490,7 +1522,10 @@ mod tests {
buffer_b.read_with(&cx_b, |buf, _| assert!(!buf.is_dirty()));
buffer_c.condition(&cx_c, |buf, _| !buf.is_dirty()).await;
// Make changes on host's file system, see those changes on the guests.
// Make changes on host's file system, see those changes on guest worktrees.
fs.rename("/a/file1".as_ref(), "/a/file1-renamed".as_ref())
.await
.unwrap();
fs.rename("/a/file2".as_ref(), "/a/file3".as_ref())
.await
.unwrap();
@@ -1498,18 +1533,29 @@ mod tests {
.await
.unwrap();
worktree_a
.condition(&cx_a, |tree, _| tree.file_count() == 4)
.await;
worktree_b
.condition(&cx_b, |tree, _| tree.file_count() == 4)
.await;
worktree_c
.condition(&cx_c, |tree, _| tree.file_count() == 4)
.await;
worktree_a.read_with(&cx_a, |tree, _| {
assert_eq!(
tree.paths()
.map(|p| p.to_string_lossy())
.collect::<Vec<_>>(),
&[".zed.toml", "file1-renamed", "file3", "file4"]
)
});
worktree_b.read_with(&cx_b, |tree, _| {
assert_eq!(
tree.paths()
.map(|p| p.to_string_lossy())
.collect::<Vec<_>>(),
&[".zed.toml", "file1", "file3", "file4"]
&[".zed.toml", "file1-renamed", "file3", "file4"]
)
});
worktree_c.read_with(&cx_c, |tree, _| {
@@ -1517,9 +1563,26 @@ mod tests {
tree.paths()
.map(|p| p.to_string_lossy())
.collect::<Vec<_>>(),
&[".zed.toml", "file1", "file3", "file4"]
&[".zed.toml", "file1-renamed", "file3", "file4"]
)
});
// Ensure buffer files are updated as well.
buffer_a
.condition(&cx_a, |buf, _| {
buf.file().unwrap().path().to_str() == Some("file1-renamed")
})
.await;
buffer_b
.condition(&cx_b, |buf, _| {
buf.file().unwrap().path().to_str() == Some("file1-renamed")
})
.await;
buffer_c
.condition(&cx_c, |buf, _| {
buf.file().unwrap().path().to_str() == Some("file1-renamed")
})
.await;
}
#[gpui::test]
@@ -1615,6 +1678,88 @@ mod tests {
});
}
#[gpui::test]
async fn test_buffer_reloading(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
cx_a.foreground().forbid_parking();
let lang_registry = Arc::new(LanguageRegistry::new());
let fs = Arc::new(FakeFs::new());
// Connect to a server as 2 clients.
let mut server = TestServer::start(cx_a.foreground()).await;
let client_a = server.create_client(&mut cx_a, "user_a").await;
let client_b = server.create_client(&mut cx_b, "user_b").await;
// Share a project as client A
fs.insert_tree(
"/dir",
json!({
".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
"a.txt": "a-contents",
}),
)
.await;
let project_a = cx_a.update(|cx| {
Project::local(
client_a.clone(),
client_a.user_store.clone(),
lang_registry.clone(),
fs.clone(),
cx,
)
});
let (worktree_a, _) = project_a
.update(&mut cx_a, |p, cx| {
p.find_or_create_local_worktree("/dir", false, cx)
})
.await
.unwrap();
worktree_a
.read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
.await;
let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
project_a
.update(&mut cx_a, |p, cx| p.share(cx))
.await
.unwrap();
// Join that project as client B
let project_b = Project::remote(
project_id,
client_b.clone(),
client_b.user_store.clone(),
lang_registry.clone(),
fs.clone(),
&mut cx_b.to_async(),
)
.await
.unwrap();
let _worktree_b = project_b.update(&mut cx_b, |p, cx| p.worktrees(cx).next().unwrap());
// Open a buffer as client B
let buffer_b = project_b
.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
.await
.unwrap();
buffer_b.read_with(&cx_b, |buf, _| {
assert!(!buf.is_dirty());
assert!(!buf.has_conflict());
});
fs.save(Path::new("/dir/a.txt"), &"new contents".into())
.await
.unwrap();
buffer_b
.condition(&cx_b, |buf, _| {
buf.text() == "new contents" && !buf.is_dirty()
})
.await;
buffer_b.read_with(&cx_b, |buf, _| {
assert!(!buf.has_conflict());
});
}
#[gpui::test]
async fn test_editing_while_guest_opens_buffer(
mut cx_a: TestAppContext,