use crate::{ copy_recursive, ignore::IgnoreStack, DiagnosticSummary, ProjectEntryId, RemoveOptions,};use ::ignore::gitignore::{Gitignore, GitignoreBuilder};use anyhow::{anyhow, Context, Result};use client::{proto, Client};use clock::ReplicaId;use collections::{HashMap, HashSet, VecDeque};use fs::{ repository::{GitFileStatus, GitRepository, RepoPath}, Fs,};use futures::{ channel::{ mpsc::{self, UnboundedSender}, oneshot, }, select_biased, task::Poll, FutureExt, Stream, StreamExt,};use fuzzy::CharBag;use git::{DOT_GIT, GITIGNORE};use gpui::{executor, AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, Task};use language::{ proto::{ deserialize_fingerprint, deserialize_version, serialize_fingerprint, serialize_line_ending, serialize_version, }, Buffer, DiagnosticEntry, File as _, LineEnding, PointUtf16, Rope, RopeFingerprint, Unclipped,};use lsp::LanguageServerId;use parking_lot::Mutex;use postage::{ barrier, prelude::{Sink as _, Stream as _}, watch,};use smol::channel::{self, Sender};use std::{ any::Any, cmp::{self, Ordering}, convert::TryFrom, ffi::OsStr, fmt, future::Future, mem, ops::{AddAssign, Deref, DerefMut, Sub}, path::{Path, PathBuf}, pin::Pin, sync::{ atomic::{AtomicUsize, Ordering::SeqCst}, Arc, }, time::{Duration, SystemTime},};use sum_tree::{Bias, Edit, SeekTarget, SumTree, TreeMap, TreeSet};use util::{paths::HOME, ResultExt}; #[derive(Copy, Clone, PartialEq, Eq, Debug, Hash, PartialOrd, Ord)]pub struct WorktreeId(usize); pub enum Worktree { Local(LocalWorktree), Remote(RemoteWorktree),} pub struct LocalWorktree { snapshot: LocalSnapshot, scan_requests_tx: channel::Sender<ScanRequest>, path_prefixes_to_scan_tx: channel::Sender<Arc<Path>>, is_scanning: (watch::Sender<bool>, watch::Receiver<bool>), _background_scanner_task: Task<()>, share: Option<ShareState>, diagnostics: HashMap< Arc<Path>, Vec<( LanguageServerId, Vec<DiagnosticEntry<Unclipped<PointUtf16>>>, )>, >, diagnostic_summaries: HashMap<Arc<Path>, HashMap<LanguageServerId, DiagnosticSummary>>, client: Arc<Client>, fs: Arc<dyn Fs>, visible: bool,} struct ScanRequest { relative_paths: Vec<Arc<Path>>, done: barrier::Sender,} pub struct RemoteWorktree { snapshot: Snapshot, background_snapshot: Arc<Mutex<Snapshot>>, project_id: u64, client: Arc<Client>, updates_tx: Option<UnboundedSender<proto::UpdateWorktree>>, snapshot_subscriptions: VecDeque<(usize, oneshot::Sender<()>)>, replica_id: ReplicaId, diagnostic_summaries: HashMap<Arc<Path>, HashMap<LanguageServerId, DiagnosticSummary>>, visible: bool, disconnected: bool,} #[derive(Clone)]pub struct Snapshot { id: WorktreeId, abs_path: Arc<Path>, root_name: String, root_char_bag: CharBag, entries_by_path: SumTree<Entry>, entries_by_id: SumTree<PathEntry>, repository_entries: TreeMap<RepositoryWorkDirectory, RepositoryEntry>, /// A number that increases every time the worktree begins scanning /// a set of paths from the filesystem. This scanning could be caused /// by some operation performed on the worktree, such as reading or /// writing a file, or by an event reported by the filesystem. scan_id: usize, /// The latest scan id that has completed, and whose preceding scans /// have all completed. The current `scan_id` could be more than one /// greater than the `completed_scan_id` if operations are performed /// on the worktree while it is processing a file-system event. completed_scan_id: usize,} #[derive(Clone, Debug, PartialEq, Eq)]pub struct RepositoryEntry { pub(crate) work_directory: WorkDirectoryEntry, pub(crate) branch: Option<Arc<str>>,} impl RepositoryEntry { pub fn branch(&self) -> Option<Arc<str>> { self.branch.clone() } pub fn work_directory_id(&self) -> ProjectEntryId { *self.work_directory } pub fn work_directory(&self, snapshot: &Snapshot) -> Option<RepositoryWorkDirectory> { snapshot .entry_for_id(self.work_directory_id()) .map(|entry| RepositoryWorkDirectory(entry.path.clone())) } pub fn build_update(&self, _: &Self) -> proto::RepositoryEntry { proto::RepositoryEntry { work_directory_id: self.work_directory_id().to_proto(), branch: self.branch.as_ref().map(|str| str.to_string()), } }} impl From<&RepositoryEntry> for proto::RepositoryEntry { fn from(value: &RepositoryEntry) -> Self { proto::RepositoryEntry { work_directory_id: value.work_directory.to_proto(), branch: value.branch.as_ref().map(|str| str.to_string()), } }} /// This path corresponds to the 'content path' (the folder that contains the .git)#[derive(Clone, Debug, Ord, PartialOrd, Eq, PartialEq)]pub struct RepositoryWorkDirectory(pub(crate) Arc<Path>); impl Default for RepositoryWorkDirectory { fn default() -> Self { RepositoryWorkDirectory(Arc::from(Path::new(""))) }} impl AsRef<Path> for RepositoryWorkDirectory { fn as_ref(&self) -> &Path { self.0.as_ref() }} #[derive(Clone, Debug, Ord, PartialOrd, Eq, PartialEq)]pub struct WorkDirectoryEntry(ProjectEntryId); impl WorkDirectoryEntry { pub(crate) fn relativize(&self, worktree: &Snapshot, path: &Path) -> Option<RepoPath> { worktree.entry_for_id(self.0).and_then(|entry| { path.strip_prefix(&entry.path) .ok() .map(move |path| path.into()) }) }} impl Deref for WorkDirectoryEntry { type Target = ProjectEntryId; fn deref(&self) -> &Self::Target { &self.0 }} impl<'a> From<ProjectEntryId> for WorkDirectoryEntry { fn from(value: ProjectEntryId) -> Self { WorkDirectoryEntry(value) }} #[derive(Debug, Clone)]pub struct LocalSnapshot { snapshot: Snapshot, /// All of the gitignore files in the worktree, indexed by their relative path. /// The boolean indicates whether the gitignore needs to be updated. ignores_by_parent_abs_path: HashMap<Arc<Path>, (Arc<Gitignore>, bool)>, /// All of the git repositories in the worktree, indexed by the project entry /// id of their parent directory. git_repositories: TreeMap<ProjectEntryId, LocalRepositoryEntry>,} struct BackgroundScannerState { snapshot: LocalSnapshot, scanned_dirs: HashSet<ProjectEntryId>, path_prefixes_to_scan: HashSet<Arc<Path>>, paths_to_scan: HashSet<Arc<Path>>, /// The ids of all of the entries that were removed from the snapshot /// as part of the current update. These entry ids may be re-used /// if the same inode is discovered at a new path, or if the given /// path is re-created after being deleted. removed_entry_ids: HashMap<u64, ProjectEntryId>, changed_paths: Vec<Arc<Path>>, prev_snapshot: Snapshot,} #[derive(Debug, Clone)]pub struct LocalRepositoryEntry { pub(crate) git_dir_scan_id: usize, pub(crate) repo_ptr: Arc<Mutex<dyn GitRepository>>, /// Path to the actual .git folder. /// Note: if .git is a file, this points to the folder indicated by the .git file pub(crate) git_dir_path: Arc<Path>,} 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 }} enum ScanState { Started, Updated { snapshot: LocalSnapshot, changes: UpdatedEntriesSet, barrier: Option<barrier::Sender>, scanning: bool, },} struct ShareState { project_id: u64, snapshots_tx: mpsc::UnboundedSender<(LocalSnapshot, UpdatedEntriesSet, UpdatedGitRepositoriesSet)>, resume_updates: watch::Sender<()>, _maintain_remote_snapshot: Task<Option<()>>,} pub enum Event { UpdatedEntries(UpdatedEntriesSet), UpdatedGitRepositories(UpdatedGitRepositoriesSet),} impl Entity for Worktree { type Event = Event;} impl Worktree { pub async fn local( client: Arc<Client>, path: impl Into<Arc<Path>>, visible: bool, fs: Arc<dyn Fs>, next_entry_id: Arc<AtomicUsize>, cx: &mut AsyncAppContext, ) -> Result<ModelHandle<Self>> { // After determining whether the root entry is a file or a directory, populate the // snapshot's "root name", which will be used for the purpose of fuzzy matching. let abs_path = path.into(); let metadata = fs .metadata(&abs_path) .await .context("failed to stat worktree path")?; Ok(cx.add_model(move |cx: &mut ModelContext<Worktree>| { let root_name = abs_path .file_name() .map_or(String::new(), |f| f.to_string_lossy().to_string()); let mut snapshot = LocalSnapshot { ignores_by_parent_abs_path: Default::default(), git_repositories: Default::default(), snapshot: Snapshot { id: WorktreeId::from_usize(cx.model_id()), abs_path: abs_path.clone(), root_name: root_name.clone(), root_char_bag: root_name.chars().map(|c| c.to_ascii_lowercase()).collect(), entries_by_path: Default::default(), entries_by_id: Default::default(), repository_entries: Default::default(), scan_id: 1, completed_scan_id: 0, }, }; if let Some(metadata) = metadata { snapshot.insert_entry( Entry::new( Arc::from(Path::new("")), &metadata, &next_entry_id, snapshot.root_char_bag, ), fs.as_ref(), ); } let (scan_requests_tx, scan_requests_rx) = channel::unbounded(); let (path_prefixes_to_scan_tx, path_prefixes_to_scan_rx) = channel::unbounded(); let (scan_states_tx, mut scan_states_rx) = mpsc::unbounded(); cx.spawn_weak(|this, mut cx| async move { while let Some((state, this)) = scan_states_rx.next().await.zip(this.upgrade(&cx)) { this.update(&mut cx, |this, cx| { let this = this.as_local_mut().unwrap(); match state { ScanState::Started => { *this.is_scanning.0.borrow_mut() = true; } ScanState::Updated { snapshot, changes, barrier, scanning, } => { *this.is_scanning.0.borrow_mut() = scanning; this.set_snapshot(snapshot, changes, cx); drop(barrier); } } cx.notify(); }); } }) .detach(); let background_scanner_task = cx.background().spawn({ let fs = fs.clone(); let snapshot = snapshot.clone(); let background = cx.background().clone(); async move { let events = fs.watch(&abs_path, Duration::from_millis(100)).await; BackgroundScanner::new( snapshot, next_entry_id, fs, scan_states_tx, background, scan_requests_rx, path_prefixes_to_scan_rx, ) .run(events) .await; } }); Worktree::Local(LocalWorktree { snapshot, is_scanning: watch::channel_with(true), share: None, scan_requests_tx, path_prefixes_to_scan_tx, _background_scanner_task: background_scanner_task, diagnostics: Default::default(), diagnostic_summaries: Default::default(), client, fs, visible, }) })) } // abcdefghi pub fn remote( project_remote_id: u64, replica_id: ReplicaId, worktree: proto::WorktreeMetadata, client: Arc<Client>, cx: &mut AppContext, ) -> ModelHandle<Self> { cx.add_model(|cx: &mut ModelContext<Self>| { let snapshot = Snapshot { id: WorktreeId(worktree.id as usize), abs_path: Arc::from(PathBuf::from(worktree.abs_path)), root_name: worktree.root_name.clone(), root_char_bag: worktree .root_name .chars() .map(|c| c.to_ascii_lowercase()) .collect(), entries_by_path: Default::default(), entries_by_id: Default::default(), repository_entries: Default::default(), scan_id: 1, completed_scan_id: 0, }; let (updates_tx, mut updates_rx) = mpsc::unbounded(); let background_snapshot = Arc::new(Mutex::new(snapshot.clone())); let (mut snapshot_updated_tx, mut snapshot_updated_rx) = watch::channel(); cx.background() .spawn({ let background_snapshot = background_snapshot.clone(); async move { while let Some(update) = updates_rx.next().await { if let Err(error) = background_snapshot.lock().apply_remote_update(update) { log::error!("error applying worktree update: {}", error); } snapshot_updated_tx.send(()).await.ok(); } } }) .detach(); cx.spawn_weak(|this, mut cx| async move { while (snapshot_updated_rx.recv().await).is_some() { if let Some(this) = this.upgrade(&cx) { this.update(&mut cx, |this, cx| { let this = this.as_remote_mut().unwrap(); this.snapshot = this.background_snapshot.lock().clone(); cx.emit(Event::UpdatedEntries(Arc::from([]))); cx.notify(); while let Some((scan_id, _)) = this.snapshot_subscriptions.front() { if this.observed_snapshot(*scan_id) { let (_, tx) = this.snapshot_subscriptions.pop_front().unwrap(); let _ = tx.send(()); } else { break; } } }); } else { break; } } }) .detach(); Worktree::Remote(RemoteWorktree { project_id: project_remote_id, replica_id, snapshot: snapshot.clone(), background_snapshot, updates_tx: Some(updates_tx), snapshot_subscriptions: Default::default(), client: client.clone(), diagnostic_summaries: Default::default(), visible: worktree.visible, disconnected: false, }) }) } pub fn as_local(&self) -> Option<&LocalWorktree> { if let Worktree::Local(worktree) = self { Some(worktree) } else { None } } pub fn as_remote(&self) -> Option<&RemoteWorktree> { if let Worktree::Remote(worktree) = self { Some(worktree) } else { None } } pub fn as_local_mut(&mut self) -> Option<&mut LocalWorktree> {Lines 1–500 of 4368. Load more
use crate::{ copy_recursive, ignore::IgnoreStack, DiagnosticSummary, ProjectEntryId, RemoveOptions, }; use ::ignore::gitignore::{Gitignore, GitignoreBuilder}; use anyhow::{anyhow, Context, Result}; use client::{proto, Client}; use clock::ReplicaId; use collections::{HashMap, HashSet, VecDeque}; use fs::{ repository::{GitFileStatus, GitRepository, RepoPath}, Fs, }; use futures::{ channel::{ mpsc::{self, UnboundedSender}, oneshot, }, select_biased, task::Poll, FutureExt, Stream, StreamExt, }; use fuzzy::CharBag; use git::{DOT_GIT, GITIGNORE}; use gpui::{executor, AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, Task}; use language::{ proto::{ deserialize_fingerprint, deserialize_version, serialize_fingerprint, serialize_line_ending, serialize_version, }, Buffer, DiagnosticEntry, File as _, LineEnding, PointUtf16, Rope, RopeFingerprint, Unclipped, }; use lsp::LanguageServerId; use parking_lot::Mutex; use postage::{ barrier, prelude::{Sink as _, Stream as _}, watch, }; use smol::channel::{self, Sender}; use std::{ any::Any, cmp::{self, Ordering}, convert::TryFrom, ffi::OsStr, fmt, future::Future, mem, ops::{AddAssign, Deref, DerefMut, Sub}, path::{Path, PathBuf}, pin::Pin, sync::{ atomic::{AtomicUsize, Ordering::SeqCst}, Arc, }, time::{Duration, SystemTime}, }; use sum_tree::{Bias, Edit, SeekTarget, SumTree, TreeMap, TreeSet}; use util::{paths::HOME, ResultExt}; #[derive(Copy, Clone, PartialEq, Eq, Debug, Hash, PartialOrd, Ord)] pub struct WorktreeId(usize); pub enum Worktree { Local(LocalWorktree), Remote(RemoteWorktree), } pub struct LocalWorktree { snapshot: LocalSnapshot, scan_requests_tx: channel::Sender<ScanRequest>, path_prefixes_to_scan_tx: channel::Sender<Arc<Path>>, is_scanning: (watch::Sender<bool>, watch::Receiver<bool>), _background_scanner_task: Task<()>, share: Option<ShareState>, diagnostics: HashMap< Arc<Path>, Vec<( LanguageServerId, Vec<DiagnosticEntry<Unclipped<PointUtf16>>>, )>, >, diagnostic_summaries: HashMap<Arc<Path>, HashMap<LanguageServerId, DiagnosticSummary>>, client: Arc<Client>, fs: Arc<dyn Fs>, visible: bool, } struct ScanRequest { relative_paths: Vec<Arc<Path>>, done: barrier::Sender, } pub struct RemoteWorktree { snapshot: Snapshot, background_snapshot: Arc<Mutex<Snapshot>>, project_id: u64, client: Arc<Client>, updates_tx: Option<UnboundedSender<proto::UpdateWorktree>>, snapshot_subscriptions: VecDeque<(usize, oneshot::Sender<()>)>, replica_id: ReplicaId, diagnostic_summaries: HashMap<Arc<Path>, HashMap<LanguageServerId, DiagnosticSummary>>, visible: bool, disconnected: bool, } #[derive(Clone)] pub struct Snapshot { id: WorktreeId, abs_path: Arc<Path>, root_name: String, root_char_bag: CharBag, entries_by_path: SumTree<Entry>, entries_by_id: SumTree<PathEntry>, repository_entries: TreeMap<RepositoryWorkDirectory, RepositoryEntry>, /// A number that increases every time the worktree begins scanning /// a set of paths from the filesystem. This scanning could be caused /// by some operation performed on the worktree, such as reading or /// writing a file, or by an event reported by the filesystem. scan_id: usize, /// The latest scan id that has completed, and whose preceding scans /// have all completed. The current `scan_id` could be more than one /// greater than the `completed_scan_id` if operations are performed /// on the worktree while it is processing a file-system event. completed_scan_id: usize, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RepositoryEntry { pub(crate) work_directory: WorkDirectoryEntry, pub(crate) branch: Option<Arc<str>>, } impl RepositoryEntry { pub fn branch(&self) -> Option<Arc<str>> { self.branch.clone() } pub fn work_directory_id(&self) -> ProjectEntryId { *self.work_directory } pub fn work_directory(&self, snapshot: &Snapshot) -> Option<RepositoryWorkDirectory> { snapshot .entry_for_id(self.work_directory_id()) .map(|entry| RepositoryWorkDirectory(entry.path.clone())) } pub fn build_update(&self, _: &Self) -> proto::RepositoryEntry { proto::RepositoryEntry { work_directory_id: self.work_directory_id().to_proto(), branch: self.branch.as_ref().map(|str| str.to_string()), } } } impl From<&RepositoryEntry> for proto::RepositoryEntry { fn from(value: &RepositoryEntry) -> Self { proto::RepositoryEntry { work_directory_id: value.work_directory.to_proto(), branch: value.branch.as_ref().map(|str| str.to_string()), } } } /// This path corresponds to the 'content path' (the folder that contains the .git) #[derive(Clone, Debug, Ord, PartialOrd, Eq, PartialEq)] pub struct RepositoryWorkDirectory(pub(crate) Arc<Path>); impl Default for RepositoryWorkDirectory { fn default() -> Self { RepositoryWorkDirectory(Arc::from(Path::new(""))) } } impl AsRef<Path> for RepositoryWorkDirectory { fn as_ref(&self) -> &Path { self.0.as_ref() } } #[derive(Clone, Debug, Ord, PartialOrd, Eq, PartialEq)] pub struct WorkDirectoryEntry(ProjectEntryId); impl WorkDirectoryEntry { pub(crate) fn relativize(&self, worktree: &Snapshot, path: &Path) -> Option<RepoPath> { worktree.entry_for_id(self.0).and_then(|entry| { path.strip_prefix(&entry.path) .ok() .map(move |path| path.into()) }) } } impl Deref for WorkDirectoryEntry { type Target = ProjectEntryId; fn deref(&self) -> &Self::Target { &self.0 } } impl<'a> From<ProjectEntryId> for WorkDirectoryEntry { fn from(value: ProjectEntryId) -> Self { WorkDirectoryEntry(value) } } #[derive(Debug, Clone)] pub struct LocalSnapshot { snapshot: Snapshot, /// All of the gitignore files in the worktree, indexed by their relative path. /// The boolean indicates whether the gitignore needs to be updated. ignores_by_parent_abs_path: HashMap<Arc<Path>, (Arc<Gitignore>, bool)>, /// All of the git repositories in the worktree, indexed by the project entry /// id of their parent directory. git_repositories: TreeMap<ProjectEntryId, LocalRepositoryEntry>, } struct BackgroundScannerState { snapshot: LocalSnapshot, scanned_dirs: HashSet<ProjectEntryId>, path_prefixes_to_scan: HashSet<Arc<Path>>, paths_to_scan: HashSet<Arc<Path>>, /// The ids of all of the entries that were removed from the snapshot /// as part of the current update. These entry ids may be re-used /// if the same inode is discovered at a new path, or if the given /// path is re-created after being deleted. removed_entry_ids: HashMap<u64, ProjectEntryId>, changed_paths: Vec<Arc<Path>>, prev_snapshot: Snapshot, } #[derive(Debug, Clone)] pub struct LocalRepositoryEntry { pub(crate) git_dir_scan_id: usize, pub(crate) repo_ptr: Arc<Mutex<dyn GitRepository>>, /// Path to the actual .git folder. /// Note: if .git is a file, this points to the folder indicated by the .git file pub(crate) git_dir_path: Arc<Path>, } 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 } } enum ScanState { Started, Updated { snapshot: LocalSnapshot, changes: UpdatedEntriesSet, barrier: Option<barrier::Sender>, scanning: bool, }, } struct ShareState { project_id: u64, snapshots_tx: mpsc::UnboundedSender<(LocalSnapshot, UpdatedEntriesSet, UpdatedGitRepositoriesSet)>, resume_updates: watch::Sender<()>, _maintain_remote_snapshot: Task<Option<()>>, } pub enum Event { UpdatedEntries(UpdatedEntriesSet), UpdatedGitRepositories(UpdatedGitRepositoriesSet), } impl Entity for Worktree { type Event = Event; } impl Worktree { pub async fn local( client: Arc<Client>, path: impl Into<Arc<Path>>, visible: bool, fs: Arc<dyn Fs>, next_entry_id: Arc<AtomicUsize>, cx: &mut AsyncAppContext, ) -> Result<ModelHandle<Self>> { // After determining whether the root entry is a file or a directory, populate the // snapshot's "root name", which will be used for the purpose of fuzzy matching. let abs_path = path.into(); let metadata = fs .metadata(&abs_path) .await .context("failed to stat worktree path")?; Ok(cx.add_model(move |cx: &mut ModelContext<Worktree>| { let root_name = abs_path .file_name() .map_or(String::new(), |f| f.to_string_lossy().to_string()); let mut snapshot = LocalSnapshot { ignores_by_parent_abs_path: Default::default(), git_repositories: Default::default(), snapshot: Snapshot { id: WorktreeId::from_usize(cx.model_id()), abs_path: abs_path.clone(), root_name: root_name.clone(), root_char_bag: root_name.chars().map(|c| c.to_ascii_lowercase()).collect(), entries_by_path: Default::default(), entries_by_id: Default::default(), repository_entries: Default::default(), scan_id: 1, completed_scan_id: 0, }, }; if let Some(metadata) = metadata { snapshot.insert_entry( Entry::new( Arc::from(Path::new("")), &metadata, &next_entry_id, snapshot.root_char_bag, ), fs.as_ref(), ); } let (scan_requests_tx, scan_requests_rx) = channel::unbounded(); let (path_prefixes_to_scan_tx, path_prefixes_to_scan_rx) = channel::unbounded(); let (scan_states_tx, mut scan_states_rx) = mpsc::unbounded(); cx.spawn_weak(|this, mut cx| async move { while let Some((state, this)) = scan_states_rx.next().await.zip(this.upgrade(&cx)) { this.update(&mut cx, |this, cx| { let this = this.as_local_mut().unwrap(); match state { ScanState::Started => { *this.is_scanning.0.borrow_mut() = true; } ScanState::Updated { snapshot, changes, barrier, scanning, } => { *this.is_scanning.0.borrow_mut() = scanning; this.set_snapshot(snapshot, changes, cx); drop(barrier); } } cx.notify(); }); } }) .detach(); let background_scanner_task = cx.background().spawn({ let fs = fs.clone(); let snapshot = snapshot.clone(); let background = cx.background().clone(); async move { let events = fs.watch(&abs_path, Duration::from_millis(100)).await; BackgroundScanner::new( snapshot, next_entry_id, fs, scan_states_tx, background, scan_requests_rx, path_prefixes_to_scan_rx, ) .run(events) .await; } }); Worktree::Local(LocalWorktree { snapshot, is_scanning: watch::channel_with(true), share: None, scan_requests_tx, path_prefixes_to_scan_tx, _background_scanner_task: background_scanner_task, diagnostics: Default::default(), diagnostic_summaries: Default::default(), client, fs, visible, }) })) } // abcdefghi pub fn remote( project_remote_id: u64, replica_id: ReplicaId, worktree: proto::WorktreeMetadata, client: Arc<Client>, cx: &mut AppContext, ) -> ModelHandle<Self> { cx.add_model(|cx: &mut ModelContext<Self>| { let snapshot = Snapshot { id: WorktreeId(worktree.id as usize), abs_path: Arc::from(PathBuf::from(worktree.abs_path)), root_name: worktree.root_name.clone(), root_char_bag: worktree .root_name .chars() .map(|c| c.to_ascii_lowercase()) .collect(), entries_by_path: Default::default(), entries_by_id: Default::default(), repository_entries: Default::default(), scan_id: 1, completed_scan_id: 0, }; let (updates_tx, mut updates_rx) = mpsc::unbounded(); let background_snapshot = Arc::new(Mutex::new(snapshot.clone())); let (mut snapshot_updated_tx, mut snapshot_updated_rx) = watch::channel(); cx.background() .spawn({ let background_snapshot = background_snapshot.clone(); async move { while let Some(update) = updates_rx.next().await { if let Err(error) = background_snapshot.lock().apply_remote_update(update) { log::error!("error applying worktree update: {}", error); } snapshot_updated_tx.send(()).await.ok(); } } }) .detach(); cx.spawn_weak(|this, mut cx| async move { while (snapshot_updated_rx.recv().await).is_some() { if let Some(this) = this.upgrade(&cx) { this.update(&mut cx, |this, cx| { let this = this.as_remote_mut().unwrap(); this.snapshot = this.background_snapshot.lock().clone(); cx.emit(Event::UpdatedEntries(Arc::from([]))); cx.notify(); while let Some((scan_id, _)) = this.snapshot_subscriptions.front() { if this.observed_snapshot(*scan_id) { let (_, tx) = this.snapshot_subscriptions.pop_front().unwrap(); let _ = tx.send(()); } else { break; } } }); } else { break; } } }) .detach(); Worktree::Remote(RemoteWorktree { project_id: project_remote_id, replica_id, snapshot: snapshot.clone(), background_snapshot, updates_tx: Some(updates_tx), snapshot_subscriptions: Default::default(), client: client.clone(), diagnostic_summaries: Default::default(), visible: worktree.visible, disconnected: false, }) }) } pub fn as_local(&self) -> Option<&LocalWorktree> { if let Worktree::Local(worktree) = self { Some(worktree) } else { None } } pub fn as_remote(&self) -> Option<&RemoteWorktree> { if let Worktree::Remote(worktree) = self { Some(worktree) } else { None } } pub fn as_local_mut(&mut self) -> Option<&mut LocalWorktree> {