Showing 4 changed files with 116 additions and 21 deletions.
| @@ -7,11 +7,12 @@ | ||
| 7 | 7 | //! slower one it keeps trying the faster ones and moves as soon as one works. |
| 8 | 8 | |
| 9 | 9 | use std::{ |
| 10 | + collections::HashMap, | |
| 10 | 11 | io::ErrorKind, |
| 11 | 12 | net::{TcpStream, ToSocketAddrs}, |
| 12 | 13 | sync::{ |
| 13 | 14 | Arc, Mutex, |
| 14 | - atomic::{AtomicBool, Ordering}, | |
| 15 | + atomic::{AtomicBool, AtomicU64, Ordering}, | |
| 15 | 16 | mpsc, |
| 16 | 17 | }, |
| 17 | 18 | thread, |
| @@ -20,7 +21,9 @@ use std::{ | ||
| 20 | 21 | |
| 21 | 22 | use tungstenite::{Message, WebSocket}; |
| 22 | 23 | |
| 23 | - use super::{ClientMessage, DEFAULT_PORT, PROTOCOL, Route, RouteKind, ServerMessage, tailnet_peers, tailnet_routes}; | |
| 24 | + use super::{ | |
| 25 | + ClientMessage, DEFAULT_PORT, PROTOCOL, RemoteFileDiff, Route, RouteKind, ServerMessage, tailnet_peers, tailnet_routes, | |
| 26 | + }; | |
| 24 | 27 | |
| 25 | 28 | /// How long to wait for a route that goes over the internet or a VPN. |
| 26 | 29 | const CONNECT_TIMEOUT: Duration = Duration::from_secs(6); |
| @@ -65,6 +68,9 @@ pub struct RemoteClient { | ||
| 65 | 68 | stop: Arc<AtomicBool>, |
| 66 | 69 | routes: Arc<Mutex<Vec<Route>>>, |
| 67 | 70 | pub events: smol::channel::Receiver<ClientEvent>, |
| 71 | + /// Diffs asked for and not answered yet, by request. | |
| 72 | + diffs: Mutex<HashMap<u64, smol::channel::Sender<Result<Vec<RemoteFileDiff>, String>>>>, | |
| 73 | + next_request: AtomicU64, | |
| 68 | 74 | } |
| 69 | 75 | |
| 70 | 76 | /// A way to the host that works off the UI thread. |
| @@ -138,7 +144,38 @@ impl RemoteClient { | ||
| 138 | 144 | } |
| 139 | 145 | }); |
| 140 | 146 | } |
| 141 | - Self { outbox, stop, routes, events } | |
| 147 | + Self { outbox, stop, routes, events, diffs: Mutex::default(), next_request: AtomicU64::new(1) } | |
| 148 | + } | |
| 149 | + | |
| 150 | + /// Asks the host for the diff between two of a thread's snapshots. The | |
| 151 | + /// answer arrives on the returned channel once [`Self::resolve_diff`] | |
| 152 | + /// gets it; older hosts never answer. | |
| 153 | + pub fn request_diff( | |
| 154 | + &self, | |
| 155 | + thread: String, | |
| 156 | + base: String, | |
| 157 | + head: String, | |
| 158 | + path: Option<String>, | |
| 159 | + ignore_whitespace: bool, | |
| 160 | + ) -> (u64, smol::channel::Receiver<Result<Vec<RemoteFileDiff>, String>>) { | |
| 161 | + let request = self.next_request.fetch_add(1, Ordering::Relaxed); | |
| 162 | + let (tx, rx) = smol::channel::bounded(1); | |
| 163 | + self.diffs.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).insert(request, tx); | |
| 164 | + self.send(&ClientMessage::Diff { request, thread, base, head, path, ignore_whitespace }); | |
| 165 | + (request, rx) | |
| 166 | + } | |
| 167 | + | |
| 168 | + /// Hands a diff reply to the request waiting for it. | |
| 169 | + pub fn resolve_diff(&self, request: u64, result: Result<Vec<RemoteFileDiff>, String>) { | |
| 170 | + let waiting = self.diffs.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).remove(&request); | |
| 171 | + if let Some(waiting) = waiting { | |
| 172 | + let _ = waiting.try_send(result); | |
| 173 | + } | |
| 174 | + } | |
| 175 | + | |
| 176 | + /// Forgets a request that stopped waiting. | |
| 177 | + pub fn forget_diff(&self, request: u64) { | |
| 178 | + self.diffs.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).remove(&request); | |
| 142 | 179 | } |
| 143 | 180 | |
| 144 | 181 | /// Changes the routes without dropping the connection; the next try, |
| @@ -436,6 +436,27 @@ impl From<&FileDiff> for RemoteFileDiff { | ||
| 436 | 436 | } |
| 437 | 437 | } |
| 438 | 438 | |
| 439 | + impl From<RemoteFileDiff> for FileDiff { | |
| 440 | + fn from(file: RemoteFileDiff) -> Self { | |
| 441 | + let lines = file | |
| 442 | + .lines | |
| 443 | + .into_iter() | |
| 444 | + .map(|line| crate::git::DiffLine { | |
| 445 | + kind: match line.kind.as_str() { | |
| 446 | + "hunk" => LineKind::Hunk, | |
| 447 | + "added" => LineKind::Added, | |
| 448 | + "removed" => LineKind::Removed, | |
| 449 | + _ => LineKind::Context, | |
| 450 | + }, | |
| 451 | + text: line.text, | |
| 452 | + old_line: line.old_line, | |
| 453 | + new_line: line.new_line, | |
| 454 | + }) | |
| 455 | + .collect(); | |
| 456 | + Self { path: file.path, binary: file.binary, lines } | |
| 457 | + } | |
| 458 | + } | |
| 459 | + | |
| 439 | 460 | /// What kind of network a route goes over, which says how fast it likely is. |
| 440 | 461 | #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] |
| 441 | 462 | #[serde(rename_all = "snake_case")] |
| @@ -1029,6 +1050,10 @@ mod tests { | ||
| 1029 | 1050 | ServerMessage::DiffReply { request: 6, files: received, error: None } => assert_eq!(received, files), |
| 1030 | 1051 | _ => panic!("not a diff reply"), |
| 1031 | 1052 | } |
| 1053 | + let back = crate::git::FileDiff::from(files[0].clone()); | |
| 1054 | + assert_eq!((back.path.as_str(), back.binary, back.lines.len()), ("src/a.rs", false, 2)); | |
| 1055 | + assert_eq!((back.lines[0].kind, back.lines[1].kind), (LineKind::Hunk, LineKind::Added)); | |
| 1056 | + assert_eq!((back.lines[1].text.as_str(), back.lines[1].new_line), ("new", Some(1))); | |
| 1032 | 1057 | } |
| 1033 | 1058 | |
| 1034 | 1059 | #[::core::prelude::v1::test] |
| @@ -82,6 +82,8 @@ const PROACTIVE_FILES: usize = 3; | ||
| 82 | 82 | const PROACTIVE_LINES: u64 = 50; |
| 83 | 83 | /// The most rows the `/`, `$` and `@` menus list. |
| 84 | 84 | const MENU_ROWS: usize = 40; |
| 85 | + /// How long a remote thread's diff may take to come back from its host. | |
| 86 | + const REMOTE_DIFF_TIMEOUT: Duration = Duration::from_secs(30); | |
| 85 | 87 | |
| 86 | 88 | #[derive(Clone, Copy, Debug, PartialEq, Eq)] |
| 87 | 89 | enum Picker { |
| @@ -1265,36 +1267,62 @@ impl ChatView { | ||
| 1265 | 1267 | let Some(diff) = &self.diff else { return }; |
| 1266 | 1268 | let session = self.session.read(cx); |
| 1267 | 1269 | let Some(changes) = session.turn_changes(diff.end) else { return }; |
| 1268 | - if session.is_remote() { | |
| 1269 | - // The snapshots live in the host's repository. | |
| 1270 | - let note = Err("This thread runs on another device; its diffs open there.".to_string()); | |
| 1271 | - if let Some(diff) = self.diff.as_mut() { | |
| 1272 | - diff.set_files(note); | |
| 1273 | - } | |
| 1270 | + let (base, head) = (changes.base.clone(), changes.head.clone()); | |
| 1271 | + let (end, path, ignore_whitespace) = (diff.end, diff.path.clone(), diff.ignore_whitespace); | |
| 1272 | + if let Some(remote) = &session.remote { | |
| 1273 | + // The snapshots live in the host's repository: ask it. | |
| 1274 | + let Some(thread) = remote.thread.clone() else { return }; | |
| 1275 | + let client = remote.client.clone(); | |
| 1276 | + let (request, reply) = client.request_diff(thread, base, head, path.clone(), ignore_whitespace); | |
| 1277 | + let timeout = cx.background_executor().timer(REMOTE_DIFF_TIMEOUT); | |
| 1278 | + cx.spawn_in(window, async move |this, cx| { | |
| 1279 | + let answer = smol::future::or(async { reply.recv().await.ok() }, async { | |
| 1280 | + timeout.await; | |
| 1281 | + None | |
| 1282 | + }) | |
| 1283 | + .await; | |
| 1284 | + let files = match answer { | |
| 1285 | + Some(result) => result.map(|files| files.into_iter().map(git::FileDiff::from).collect()), | |
| 1286 | + None => { | |
| 1287 | + client.forget_diff(request); | |
| 1288 | + Err("The other device didn't answer. Its Vectrum Code may be too old to share diffs, or the connection dropped." | |
| 1289 | + .to_string()) | |
| 1290 | + } | |
| 1291 | + }; | |
| 1292 | + let _ = this.update(cx, |this, cx| this.diff_loaded(end, &path, ignore_whitespace, files, cx)); | |
| 1293 | + }) | |
| 1294 | + .detach(); | |
| 1274 | 1295 | return; |
| 1275 | 1296 | } |
| 1276 | - let (base, head) = (changes.base.clone(), changes.head.clone()); | |
| 1277 | 1297 | let cwd = session.cwd.clone(); |
| 1278 | - let (end, path, ignore_whitespace) = (diff.end, diff.path.clone(), diff.ignore_whitespace); | |
| 1279 | 1298 | let task = cx.background_executor().spawn({ |
| 1280 | 1299 | let path = path.clone(); |
| 1281 | 1300 | async move { git::diff_between(&cwd, &base, &head, path.as_deref(), ignore_whitespace) } |
| 1282 | 1301 | }); |
| 1283 | 1302 | cx.spawn_in(window, async move |this, cx| { |
| 1284 | 1303 | let files = task.await; |
| 1285 | - let _ = this.update(cx, |this, cx| { | |
| 1286 | - // Unless the diff was closed, replaced or toggled meanwhile. | |
| 1287 | - if let Some(diff) = this.diff.as_mut() | |
| 1288 | - && (diff.end, &diff.path, diff.ignore_whitespace) == (end, &path, ignore_whitespace) | |
| 1289 | - { | |
| 1290 | - diff.set_files(files); | |
| 1291 | - cx.notify(); | |
| 1292 | - } | |
| 1293 | - }); | |
| 1304 | + let _ = this.update(cx, |this, cx| this.diff_loaded(end, &path, ignore_whitespace, files, cx)); | |
| 1294 | 1305 | }) |
| 1295 | 1306 | .detach(); |
| 1296 | 1307 | } |
| 1297 | 1308 | |
| 1309 | + /// Shows a diff that was read, unless it was closed, replaced or toggled meanwhile. | |
| 1310 | + fn diff_loaded( | |
| 1311 | + &mut self, | |
| 1312 | + end: usize, | |
| 1313 | + path: &Option<String>, | |
| 1314 | + ignore_whitespace: bool, | |
| 1315 | + files: Result<Vec<git::FileDiff>, String>, | |
| 1316 | + cx: &mut Context<Self>, | |
| 1317 | + ) { | |
| 1318 | + if let Some(diff) = self.diff.as_mut() | |
| 1319 | + && (diff.end, &diff.path, diff.ignore_whitespace) == (end, path, ignore_whitespace) | |
| 1320 | + { | |
| 1321 | + diff.set_files(files); | |
| 1322 | + cx.notify(); | |
| 1323 | + } | |
| 1324 | + } | |
| 1325 | + | |
| 1298 | 1326 | fn diff_action(&mut self, action: DiffAction, window: &mut Window, cx: &mut Context<Self>) { |
| 1299 | 1327 | let Some(diff) = self.diff.as_mut() else { return }; |
| 1300 | 1328 | match action { |
| @@ -847,7 +847,12 @@ impl Workspace { | ||
| 847 | 847 | ClientEvent::Message(ServerMessage::Routes { routes }) => self.learn_routes(env, &routes), |
| 848 | 848 | ClientEvent::Message(ServerMessage::Welcome { .. } | ServerMessage::Denied { .. }) => {} |
| 849 | 849 | // Only sent in reply to requests this app doesn't make. |
| 850 | - ClientEvent::Message(ServerMessage::Catalog { .. } | ServerMessage::DiffReply { .. }) => {} | |
| 850 | + ClientEvent::Message(ServerMessage::Catalog { .. }) => {} | |
| 851 | + ClientEvent::Message(ServerMessage::DiffReply { request, files, error }) => { | |
| 852 | + if let Some(environment) = self.remote.environments.iter().find(|candidate| candidate.host.id == env) { | |
| 853 | + environment.client.resolve_diff(request, error.map_or(Ok(files), Err)); | |
| 854 | + } | |
| 855 | + } | |
| 851 | 856 | ClientEvent::Message(ServerMessage::DeviceReply { request, ok, error }) => { |
| 852 | 857 | if let Some(link) = self.devices_link(env) { |
| 853 | 858 | link.resolve(request, error.map_or_else(|| Ok(ok.unwrap_or_default()), Err)); |