hl-valdemar/vamp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Wiki Activity

feat: add local and remote repository diff views

Browse Source
This commit is contained in:
Valdemar Høgsvig Lorenzen
2026-10-10 23:01:34 +02:00
parent f6da7922a1
commit 81a2bf9a02
15 changed files with 4947 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/session.rs
+245 -1
View File
@@ -2,18 +2,125 @@ use std::{
collections::BTreeMap,
io,
path::{Path, PathBuf},
sync::{
Arc,
atomic::{AtomicBool, AtomicU64, Ordering},
mpsc::{self, Receiver, SyncSender},
},
thread::{self, JoinHandle},
};
use crate::{
diff::{Operation as DiffOperation, Outcome as DiffOutcome},
editor::{Cursor, Editor, Effect, Input, Invocation, Mode},
lsp::Lsp,
plugin::{
DocumentEvent, DocumentEventKind, DocumentSnapshot, PluginEvent, PluginUpdate,
TextPosition, UpdateEffect,
},
repository::Repository,
ruby::{Config, RubyPlugin, load_config},
};
type DiffResult = (u64, Result<DiffOutcome, String>);
struct DiffWorker {
requests: Option<SyncSender<(u64, DiffOperation)>>,
results: Option<Receiver<DiffResult>>,
cancelled: Arc<AtomicBool>,
stopped: Arc<AtomicBool>,
cancelled_request: Arc<AtomicU64>,
thread: Option<JoinHandle<()>>,
pending: Option<u64>,
}
impl DiffWorker {
fn start(root: PathBuf) -> Result<Self, String> {
let (requests, incoming) = mpsc::sync_channel::<(u64, DiffOperation)>(1);
let (outgoing, results) = mpsc::sync_channel(1);
let cancelled = Arc::new(AtomicBool::new(false));
let stopped = Arc::new(AtomicBool::new(false));
let cancelled_request = Arc::new(AtomicU64::new(0));
let cancelled_id = Arc::clone(&cancelled_request);
let stop = Arc::clone(&stopped);
let cancellation = Arc::clone(&cancelled);
let thread = thread::Builder::new()
.name("vamp-diff".into())
.spawn(move || {
let mut repository = None;
while let Ok((id, operation)) = incoming.recv() {
if stop.load(Ordering::Acquire) {
break;
}
cancellation.store(false, Ordering::Release);
if stop.load(Ordering::Acquire) {
break;
}
if cancelled_id.load(Ordering::Acquire) == id {
continue;
}
let result = (|| {
if repository.is_none() {
repository = Some(Repository::open(&root, Arc::clone(&cancellation))?);
}
let repository = repository.as_mut().expect("repository opened");
match operation {
DiffOperation::Discover => {
repository.catalog().map(DiffOutcome::Catalog)
}
DiffOperation::FetchRemote(remote) => {
repository.fetch_remote(&remote).map(DiffOutcome::Catalog)
}
DiffOperation::Compare { before, after } => repository
.compare(&before, &after)
.map(DiffOutcome::Comparison),
}
})();
if stop.load(Ordering::Acquire) || outgoing.send((id, result)).is_err() {
break;
}
}
})
.map_err(|error| format!("Cannot start diff worker: {error}"))?;
Ok(Self {
requests: Some(requests),
results: Some(results),
cancelled,
stopped,
cancelled_request,
thread: Some(thread),
pending: None,
})
}
fn request(&mut self, id: u64, operation: DiffOperation) -> Result<(), String> {
self.requests
.as_ref()
.expect("live diff worker has a sender")
.try_send((id, operation))
.map_err(|error| match error {
mpsc::TrySendError::Full(_) => "Diff worker is already busy".to_owned(),
mpsc::TrySendError::Disconnected(_) => {
"Diff worker stopped unexpectedly".to_owned()
}
})?;
self.pending = Some(id);
Ok(())
}
}
impl Drop for DiffWorker {
fn drop(&mut self) {
self.stopped.store(true, Ordering::Release);
self.cancelled.store(true, Ordering::Release);
self.requests.take();
self.results.take();
if let Some(thread) = self.thread.take() {
let _ = thread.join();
}
}
}
struct TrackedDocument {
snapshot: DocumentSnapshot,
save_revision: u64,
@@ -45,6 +152,7 @@ pub struct EditorSession {
pending_hover: bool,
pending_completion: bool,
hover_owner: Option<String>,
diff_worker: Option<DiffWorker>,
}
impl EditorSession {
@@ -135,6 +243,7 @@ impl EditorSession {
pending_hover: false,
pending_completion: false,
hover_owner: None,
diff_worker: None,
};
session.synchronize_documents();
Ok(session)
@@ -152,6 +261,7 @@ impl EditorSession {
self.dispatch(invocation);
self.synchronize_documents();
}
self.start_diff_request();
}
pub fn execute(&mut self, command: &str) {
@@ -162,6 +272,82 @@ impl EditorSession {
self.dispatch(invocation);
self.synchronize_documents();
}
self.start_diff_request();
}
fn start_diff_request(&mut self) {
if self.editor.diff().is_none() {
self.diff_worker = None;
return;
}
if self.editor.diff().is_some_and(|diff| !diff.loading)
&& let Some(worker) = &mut self.diff_worker
&& let Some(id) = worker.pending.take()
{
worker.cancelled_request.store(id, Ordering::Release);
worker.cancelled.store(true, Ordering::Release);
}
let Some((id, operation)) = self.editor.take_diff_request() else {
return;
};
if self.diff_worker.is_none() {
let root = self
.editor
.path()
.and_then(Path::parent)
.filter(|path| !path.as_os_str().is_empty())
.unwrap_or_else(|| self.editor.project_root())
.to_path_buf();
match DiffWorker::start(root) {
Ok(worker) => self.diff_worker = Some(worker),
Err(error) => {
self.editor.apply_diff_result(id, Err(error));
return;
}
}
}
if let Err(error) = self
.diff_worker
.as_mut()
.expect("diff worker started")
.request(id, operation)
{
self.editor.apply_diff_result(id, Err(error));
self.diff_worker = None;
}
}
fn poll_diff(&mut self) -> bool {
self.start_diff_request();
let mut changed = false;
while let Some(worker) = self.diff_worker.as_ref() {
let result = worker
.results
.as_ref()
.expect("live diff worker has results")
.try_recv();
match result {
Ok((id, result)) => {
if let Some(worker) = &mut self.diff_worker
&& worker.pending == Some(id)
{
worker.pending = None;
}
changed |= self.editor.apply_diff_result(id, result);
}
Err(mpsc::TryRecvError::Empty) => break,
Err(mpsc::TryRecvError::Disconnected) => {
if let Some(id) = self.diff_worker.as_ref().and_then(|worker| worker.pending) {
changed |= self
.editor
.apply_diff_result(id, Err("Diff worker stopped unexpectedly".into()));
}
self.diff_worker = None;
break;
}
}
}
changed
}
fn dispatch(&mut self, invocation: Invocation) {
@@ -368,6 +554,9 @@ impl EditorSession {
request_id: Option<u64>,
source: InvocationSource,
) -> bool {
if self.editor.diff().is_some() {
return false;
}
self.editor.buffer_id() == source.buffer_id
&& self.source_is_current(source)
&& (owner != "core"
@@ -617,7 +806,10 @@ impl EditorSession {
let Some(source) = source else {
return Err("plugin response has an unknown invocation ID".into());
};
if !self.source_is_current(source) || self.editor.buffer_id() != source.buffer_id {
if !self.source_is_current(source)
|| self.editor.buffer_id() != source.buffer_id
|| self.editor.diff().is_some()
{
return Ok(false);
}
if effects
@@ -641,6 +833,7 @@ impl EditorSession {
/// Drains ready plugin work without waiting for command or server responses.
pub fn poll(&mut self) -> bool {
let mut changed = self.synchronize_documents();
changed |= self.poll_diff();
for update in self.lsp.poll() {
match self.apply_update("core", update) {
Ok(applied) => changed |= applied,
@@ -3076,4 +3269,55 @@ end
assert_eq!(invokes[0]["column"], 2);
assert_eq!(invokes[0]["revision"], 0);
}
#[test]
fn pending_plugin_edits_do_not_modify_diff_buffers_or_disable_the_plugin() {
let fixture = Fixture::new();
let source = fixture.file("buffer.txt", "original");
let started = fixture.0.join("started");
let release = fixture.0.join("release");
fixture.file(
"delayed.rb",
r#"Vamp.plugin "delayed", requires: ["text"], async: true
Vamp.command "slow" do |editor, _|
File.write(Vamp.options.fetch("started"), "ready")
sleep 0.001 until File.exist?(Vamp.options.fetch("release"))
editor.insert "late edit"
end
Vamp.command "ping" do |editor, _|
editor.message "still alive"
end
"#,
);
let config = fixture.file(
"config.rb",
&format!(
"Vamp.configure do |config|\n config.clipboard_mode = 'separate'\n config.plugin 'delayed.rb', options: {{started: {}, release: {}}}\nend\n",
serde_json::to_string(&started.to_string_lossy()).unwrap(),
serde_json::to_string(&release.to_string_lossy()).unwrap(),
),
);
let mut session =
EditorSession::open(Some(source.clone()), vec!["text".into()], Some(&config)).unwrap();
session.execute("slow");
drain_until(&mut session, |_| started.exists());
session.execute("diff");
assert!(session.editor().diff().is_some());
fs::write(&release, "release").unwrap();
// Wait for transport consumption, then assert the consumer-visible result.
drain_until(&mut session, |session| {
session
.requests
.get("delayed")
.is_none_or(|requests| requests.pending.is_empty())
});
assert_eq!(session.editor().context().text, "original");
assert_eq!(fs::read_to_string(&source).unwrap(), "original");
session.handle(Input::Escape);
session.execute("ping");
drain_until(&mut session, |session| {
session.editor().message() == "still alive"
});
assert_eq!(session.editor().context().text, "original");
}
}