@@ -5,36 +5,107 @@ use std::{
55 process:: { Child , Stdio } ,
66 sync:: mpsc:: Sender ,
77 thread,
8- time:: Duration ,
8+ time:: { Duration , Instant } ,
99} ;
1010
1111use anyhow:: { Result , anyhow} ;
1212use log:: { info, warn} ;
1313
1414use super :: path_info:: PathInfo ;
15- use crate :: { app:: config:: Config , command:: Command } ;
16-
17- pub ( super ) fn cd ( directory : & PathInfo ) -> Result < ( Vec < PathInfo > , usize ) > {
18- info ! ( "Changing directory to {directory:?}" ) ;
19- let entries = fs:: read_dir ( & directory. path ) ?;
20-
21- // Use collect to gather results, then partition into successes and failures
22- let results: Vec < Result < PathInfo > > = entries
23- . map ( |entry| {
24- entry
25- . map_err ( Into :: into)
26- . and_then ( |e| PathInfo :: try_from ( & e. path ( ) ) )
27- } )
28- . collect ( ) ;
29-
30- let ( children, errors) : ( Vec < _ > , Vec < _ > ) = results. into_iter ( ) . partition ( Result :: is_ok) ;
31-
32- let error_count = errors. len ( ) ;
33- if error_count > 0 {
34- warn ! ( "Some paths could not be read: {:?}" , errors) ;
35- }
15+ use crate :: {
16+ app:: config:: Config ,
17+ command:: { Command , progress:: CancellationToken } ,
18+ } ;
19+
20+ // Entries are streamed in batches (rather than one command per entry) so a flood
21+ // of commands cannot starve terminal input in the single FIFO command channel.
22+ // A batch flushes once it reaches CD_BATCH_SIZE or CD_FLUSH_INTERVAL elapses.
23+ const CD_BATCH_SIZE : usize = 256 ;
24+ const CD_FLUSH_INTERVAL : Duration = Duration :: from_millis ( 80 ) ;
25+
26+ /// Spawns a background thread that reads `directory` and streams its entries as
27+ /// `Command::DirectoryListing` batches, finishing with a
28+ /// `Command::DirectoryListingComplete`. `generation` tags every message so a
29+ /// superseded load (the user navigated away) can be ignored; `cancel` stops the
30+ /// walk early when that happens. Reading off the UI thread keeps navigation into
31+ /// very large directories responsive.
32+ pub ( super ) fn stream_cd (
33+ directory : PathInfo ,
34+ generation : u64 ,
35+ tx : Sender < Command > ,
36+ cancel : CancellationToken ,
37+ ) {
38+ info ! ( "Streaming directory {directory:?}" ) ;
39+ thread:: spawn ( move || {
40+ let entries = match fs:: read_dir ( & directory. path ) {
41+ Ok ( entries) => entries,
42+ Err ( error) => {
43+ let _ = tx. send ( Command :: AlertWarn ( format ! (
44+ "Failed to read directory {:?}: {error}" ,
45+ directory. path
46+ ) ) ) ;
47+ let _ = tx. send ( Command :: DirectoryListingComplete { generation } ) ;
48+ return ;
49+ }
50+ } ;
51+
52+ let mut batch: Vec < PathInfo > = Vec :: new ( ) ;
53+ let mut last_flush = Instant :: now ( ) ;
54+ let mut error_count: usize = 0 ;
55+
56+ for entry in entries {
57+ // A newer load has superseded this one: stop without sending a
58+ // completion (the newer load owns the listing now).
59+ if cancel. is_cancelled ( ) {
60+ return ;
61+ }
62+ let path = match entry {
63+ Ok ( entry) => entry. path ( ) ,
64+ Err ( error) => {
65+ warn ! ( "Could not read an entry in {:?}: {error}" , directory. path) ;
66+ error_count += 1 ;
67+ continue ;
68+ }
69+ } ;
70+ match PathInfo :: try_from ( & path) {
71+ Ok ( info) => batch. push ( info) ,
72+ Err ( error) => {
73+ warn ! ( "Could not read metadata for {path:?}: {error}" ) ;
74+ error_count += 1 ;
75+ }
76+ }
77+ if batch. len ( ) >= CD_BATCH_SIZE || last_flush. elapsed ( ) >= CD_FLUSH_INTERVAL {
78+ if !flush_listing ( & tx, & mut batch, generation) {
79+ return ; // channel closed
80+ }
81+ last_flush = Instant :: now ( ) ;
82+ }
83+ }
84+
85+ if !flush_listing ( & tx, & mut batch, generation) {
86+ return ;
87+ }
88+ if error_count > 0 {
89+ let _ = tx. send ( Command :: AlertWarn ( format ! (
90+ "{error_count} entries in {:?} could not be read" ,
91+ directory. path
92+ ) ) ) ;
93+ }
94+ let _ = tx. send ( Command :: DirectoryListingComplete { generation } ) ;
95+ } ) ;
96+ }
3697
37- Ok ( ( children. into_iter ( ) . flatten ( ) . collect ( ) , error_count) )
98+ /// Sends the accumulated batch (if any) as a single `Command::DirectoryListing`.
99+ /// Returns `false` if the channel is closed, signalling the caller to stop.
100+ fn flush_listing ( tx : & Sender < Command > , batch : & mut Vec < PathInfo > , generation : u64 ) -> bool {
101+ if batch. is_empty ( ) {
102+ return true ;
103+ }
104+ tx. send ( Command :: DirectoryListing {
105+ items : std:: mem:: take ( batch) ,
106+ generation,
107+ } )
108+ . is_ok ( )
38109}
39110
40111pub ( super ) fn open_in ( path : & PathInfo , template : & str , command_tx : Sender < Command > ) -> Result < ( ) > {
0 commit comments