use std::fs::{ self, File, OpenOptions };
use std::io::{ self, Read, Seek, SeekFrom, Write };
use std::path::{ Path, PathBuf };
use std::sync::atomic::{ AtomicBool, AtomicU64, Ordering };
use std::sync::mpsc;
use std::sync::{ Arc, Mutex, OnceLock, PoisonError };
use std::thread;
use super::{ Entry, Kind, Metadata, Mount, Progress, Reply };
type Job = Box<dyn FnOnce() + Send>;
fn run( job: impl FnOnce() + Send + 'static )
{
static WORKER: OnceLock<Option<mpsc::Sender<Job>>> = OnceLock::new();
let worker = WORKER.get_or_init( ||
{
let ( tx, rx ) = mpsc::channel::<Job>();
thread::Builder::new()
.name( "ltk-files".into() )
.spawn( move || { for job in rx { job(); } } )
.ok()
.map( |_| tx )
} );
let job: Job = Box::new( job );
let job = match worker
{
Some( tx ) => match tx.send( job ) { Ok( () ) => return, Err( mpsc::SendError( job ) ) => job },
None => job,
};
job();
}
pub fn list( path: PathBuf, reply: Reply<Vec<Entry>> )
{
run( move || reply( list_dir( &path ) ) );
}
pub fn stat( path: PathBuf, reply: Reply<Metadata> )
{
run( move || reply( fs::metadata( &path ).map( |meta| metadata( &meta ) ) ) );
}
pub fn read( path: PathBuf, range: Option<( u64, u64 )>, reply: Reply<Vec<u8>> )
{
run( move || reply( match range
{
Some( ( offset, length ) ) => read_range( &path, offset, length ),
None => fs::read( &path ),
} ) );
}
pub fn write( path: PathBuf, data: Vec<u8>, reply: Reply<()> )
{
run( move || reply( Pending::create( &path ).and_then( |mut file| { file.write( &data )?; file.commit() } ) ) );
}
pub fn create_dir( path: PathBuf, reply: Reply<()> )
{
run( move || reply( fs::create_dir( &path ) ) );
}
pub fn rename( from: PathBuf, to: PathBuf, reply: Reply<()> )
{
run( move || reply( if fs::symlink_metadata( &to ).is_ok() { Err( io::ErrorKind::AlreadyExists.into() ) } else { fs::rename( &from, &to ) } ) );
}
pub fn remove( path: PathBuf, all: bool, reply: Reply<()> )
{
run( move || reply( remove_path( &path, all ) ) );
}
pub fn mount( reply: Reply<Mount> )
{
reply( Err( io::ErrorKind::Unsupported.into() ) );
}
pub fn unmount( _path: PathBuf, reply: Reply<()> )
{
reply( Err( io::ErrorKind::Unsupported.into() ) );
}
pub type WriterHandle = Arc<Mutex<Option<Pending>>>;
pub fn open_write( path: PathBuf, reply: Reply<WriterHandle> )
{
run( move || reply( Pending::create( &path ).map( |file| Arc::new( Mutex::new( Some( file ) ) ) ) ) );
}
pub fn write_chunk( handle: &WriterHandle, chunk: Vec<u8>, reply: Reply<()> )
{
let handle = handle.clone();
run( move || reply( match lock( &handle ).as_mut()
{
Some( file ) => file.write( &chunk ),
None => Err( io::ErrorKind::BrokenPipe.into() ),
} ) );
}
pub fn close( handle: WriterHandle, reply: Reply<()> )
{
run( move ||
{
let file = lock( &handle ).take();
reply( file.map_or_else( || Err( io::ErrorKind::BrokenPipe.into() ), Pending::commit ) );
} );
}
pub fn abort( handle: WriterHandle )
{
run( move || drop( lock( &handle ).take() ) );
}
fn lock( handle: &WriterHandle ) -> std::sync::MutexGuard<'_, Option<Pending>>
{
handle.lock().unwrap_or_else( PoisonError::into_inner )
}
pub type OperationHandle = Arc<AtomicBool>;
pub fn copy( from: PathBuf, to: PathBuf, all: bool, progress: Progress, reply: Reply<()> ) -> OperationHandle
{
let cancelled = Arc::new( AtomicBool::new( false ) );
let flag = cancelled.clone();
let job = move || reply( super::copy::copy( &from, &to, all, &flag, &progress ) );
let job: Job = Box::new( job );
let job = Arc::new( Mutex::new( Some( job ) ) );
let spawned = job.clone();
let started = thread::Builder::new()
.name( "ltk-files-copy".into() )
.spawn( move || { if let Some( job ) = spawned.lock().unwrap_or_else( PoisonError::into_inner ).take() { job(); } } )
.is_ok();
if !started
{
if let Some( job ) = job.lock().unwrap_or_else( PoisonError::into_inner ).take() { run( job ); }
}
cancelled
}
pub fn cancel( handle: &OperationHandle )
{
handle.store( true, Ordering::Relaxed );
}
pub( super ) fn list_dir( path: &Path ) -> io::Result<Vec<Entry>>
{
let mut entries = Vec::new();
for entry in fs::read_dir( path )?
{
let Ok( entry ) = entry else { continue };
let Ok( meta ) = fs::metadata( entry.path() ).or_else( |_| entry.metadata() ) else { continue };
let meta = metadata( &meta );
entries.push( Entry { name: entry.file_name(), kind: meta.kind, size: meta.size, modified: meta.modified, accessed: meta.accessed } );
}
Ok( entries )
}
fn metadata( meta: &fs::Metadata ) -> Metadata
{
let kind = if meta.is_dir() { Kind::Directory } else { Kind::File };
Metadata
{
kind,
size: if kind == Kind::File { meta.len() } else { 0 },
modified: meta.modified().ok(),
accessed: meta.accessed().ok(),
}
}
pub( super ) fn read_range( path: &Path, offset: u64, length: u64 ) -> io::Result<Vec<u8>>
{
let mut file = File::open( path )?;
file.seek( SeekFrom::Start( offset ) )?;
let mut data = Vec::new();
file.take( length ).read_to_end( &mut data )?;
Ok( data )
}
pub( super ) fn remove_path( path: &Path, all: bool ) -> io::Result<()>
{
match ( fs::symlink_metadata( path )?.is_dir(), all )
{
( true, true ) => fs::remove_dir_all( path ),
( true, false ) => fs::remove_dir( path ),
( false, _ ) => fs::remove_file( path ),
}
}
pub struct Pending
{
file: Option<File>,
temp: PathBuf,
target: PathBuf,
}
impl Pending
{
pub( super ) fn create( target: &Path ) -> io::Result<Self>
{
static NEXT: AtomicU64 = AtomicU64::new( 0 );
let name = target.file_name().ok_or( io::ErrorKind::InvalidInput )?.to_string_lossy();
let temp = target.with_file_name( format!( ".{name}.{}-{}.part", std::process::id(), NEXT.fetch_add( 1, Ordering::Relaxed ) ) );
let file = OpenOptions::new().write( true ).create_new( true ).open( &temp )?;
if let Ok( meta ) = fs::metadata( target ) { let _ = file.set_permissions( meta.permissions() ); }
Ok( Self { file: Some( file ), temp, target: target.to_path_buf() } )
}
pub( super ) fn write( &mut self, chunk: &[ u8 ] ) -> io::Result<()>
{
self.file.as_mut().ok_or( io::ErrorKind::BrokenPipe )?.write_all( chunk )
}
pub( super ) fn commit( mut self ) -> io::Result<()>
{
let file = self.file.take().ok_or( io::ErrorKind::BrokenPipe )?;
file.sync_all()?;
drop( file );
fs::rename( &self.temp, &self.target )
}
}
impl Drop for Pending
{
fn drop( &mut self )
{
if self.file.take().is_some() { let _ = fs::remove_file( &self.temp ); }
}
}
#[ cfg( test ) ]
mod tests
{
use super::*;
fn scratch( name: &str ) -> PathBuf
{
let dir = std::env::temp_dir().join( format!( "ltk-files-{name}-{}", std::process::id() ) );
let _ = fs::remove_dir_all( &dir );
fs::create_dir_all( &dir ).unwrap();
dir
}
#[ test ]
fn a_pending_file_appears_only_once_committed()
{
let dir = scratch( "commit" );
let target = dir.join( "a.txt" );
let mut file = Pending::create( &target ).unwrap();
file.write( b"one " ).unwrap();
file.write( b"two" ).unwrap();
assert!( !target.exists() );
file.commit().unwrap();
assert_eq!( fs::read( &target ).unwrap(), b"one two" );
assert_eq!( fs::read_dir( &dir ).unwrap().count(), 1 );
fs::remove_dir_all( &dir ).unwrap();
}
#[ test ]
fn a_dropped_pending_file_leaves_the_old_one()
{
let dir = scratch( "abort" );
let target = dir.join( "a.txt" );
fs::write( &target, b"old" ).unwrap();
let mut file = Pending::create( &target ).unwrap();
file.write( b"new" ).unwrap();
drop( file );
assert_eq!( fs::read( &target ).unwrap(), b"old" );
assert_eq!( fs::read_dir( &dir ).unwrap().count(), 1 );
fs::remove_dir_all( &dir ).unwrap();
}
#[ test ]
fn a_listing_gives_kinds_and_sizes()
{
let dir = scratch( "list" );
fs::write( dir.join( "f" ), b"12345" ).unwrap();
fs::create_dir( dir.join( "d" ) ).unwrap();
let mut entries = list_dir( &dir ).unwrap();
entries.sort_by( |a, b| a.name.cmp( &b.name ) );
assert_eq!( entries.len(), 2 );
assert_eq!( ( entries[ 0 ].kind, entries[ 0 ].size ), ( Kind::Directory, 0 ) );
assert_eq!( ( entries[ 1 ].kind, entries[ 1 ].size ), ( Kind::File, 5 ) );
fs::remove_dir_all( &dir ).unwrap();
}
#[ test ]
fn a_range_stops_at_the_end()
{
let dir = scratch( "range" );
let file = dir.join( "f" );
fs::write( &file, b"0123456789" ).unwrap();
assert_eq!( read_range( &file, 3, 4 ).unwrap(), b"3456" );
assert_eq!( read_range( &file, 8, 4 ).unwrap(), b"89" );
assert!( read_range( &file, 20, 4 ).unwrap().is_empty() );
fs::remove_dir_all( &dir ).unwrap();
}
#[ test ]
fn remove_needs_all_for_a_full_directory()
{
let dir = scratch( "remove" );
let sub = dir.join( "d" );
fs::create_dir( &sub ).unwrap();
fs::write( sub.join( "f" ), b"x" ).unwrap();
assert!( remove_path( &sub, false ).is_err() );
remove_path( &sub, true ).unwrap();
assert!( !sub.exists() );
fs::remove_dir_all( &dir ).unwrap();
}
}