diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/db.rs | 219 | ||||
| -rw-r--r-- | src/files.rs | 132 | ||||
| -rw-r--r-- | src/main.rs | 50 |
3 files changed, 219 insertions, 182 deletions
@@ -1,16 +1,15 @@ +use crate::flac::{CURRENT_VENDOR, get_vendor}; use anyhow::{Result, anyhow}; use directories::BaseDirs; -use rusqlite::{Connection, params}; use std::{ path::{Path, PathBuf}, time::UNIX_EPOCH, }; - -use crate::flac::{CURRENT_VENDOR, get_vendor}; +use turso::{Connection, params}; const TABLE_CREATE: &str = "CREATE TABLE IF NOT EXISTS flacs (path TEXT PRIMARY KEY UNIQUE, toencode BOOLEAN NOT NULL, modtime INTEGER)"; -const ADD_ITEM: &str = "INSERT INTO flacs (path, toencode, modtime) VALUES (?1, ?2, ?3)"; -const UPDATE_ITEM: &str = "UPDATE flacs SET toencode = ?2, modtime = ?3 WHERE path = ?1"; +const ADD_FILE: &str = "INSERT INTO flacs (path, toencode, modtime) VALUES (?1, ?2, ?3)"; +const UPDATE_FILE: &str = "UPDATE flacs SET toencode = ?2, modtime = ?3 WHERE path = ?1"; const TOENCODE_PATHS: &str = "SELECT path FROM flacs WHERE toencode"; const TOENCODE_NUMBER: &str = "SELECT COUNT(*) from flacs WHERE toencode"; const CHECK_FILE: &str = "SELECT exists(SELECT 1 FROM flacs WHERE path = ?1)"; @@ -20,20 +19,25 @@ const DEDUPE_DB: &str = "DELETE FROM flacs WHERE rowid NOT IN (SELECT MAX(rowid) FROM flacs GROUP BY path)"; const GET_MODTIME: &str = "SELECT modtime FROM flacs WHERE path = ?1"; -pub(crate) fn init_connection(path: Option<&PathBuf>) -> Result<Connection> { - let conn = if let Some(file) = path { - Connection::open(file)? +pub(crate) async fn init_db(path: Option<&PathBuf>) -> Result<turso::Database> { + let db = if let Some(file) = path { + turso::Builder::new_local(file.to_str().unwrap()) + .build() + .await? } else if let Some(base_dir) = BaseDirs::new() { - let file = Path::new(base_dir.data_dir()).join("reencoder.db"); - Connection::open(file)? + let file = base_dir.data_dir().join("reencoder.db"); + turso::Builder::new_local(file.to_str().unwrap()) + .build() + .await? } else { return Err(anyhow!("Failed to locate data directory")); }; - conn.execute(TABLE_CREATE, ())?; - Ok(conn) + let conn = db.connect()?; + conn.execute(TABLE_CREATE, ()).await?; + Ok(db) } -pub(crate) fn insert_file(conn: &Connection, filename: &Path) -> Result<()> { +pub(crate) async fn insert_file(conn: &Connection, filename: &Path) -> Result<()> { let toencode = !matches!(get_vendor(filename)?.as_str(), CURRENT_VENDOR); let modtime = filename @@ -43,14 +47,15 @@ pub(crate) fn insert_file(conn: &Connection, filename: &Path) -> Result<()> { .as_secs(); conn.execute( - ADD_ITEM, + ADD_FILE, params![filename.to_str().unwrap(), toencode, modtime], - )?; + ) + .await?; Ok(()) } -pub(crate) fn update_file(conn: &Connection, filename: &Path) -> Result<()> { +pub(crate) async fn update_file(conn: &Connection, filename: &Path) -> Result<()> { let modtime = filename .metadata()? .modified()? @@ -58,70 +63,74 @@ pub(crate) fn update_file(conn: &Connection, filename: &Path) -> Result<()> { .as_secs(); conn.execute( - UPDATE_ITEM, + UPDATE_FILE, params![filename.to_str().unwrap(), false, modtime], - )?; + ) + .await?; Ok(()) } -pub(crate) fn check_file(conn: &Connection, filename: &Path) -> Result<bool> { - if conn.query_one(CHECK_FILE, params!(filename.to_str().unwrap()), |row| { - let num: bool = row.get(0)?; - Ok(num) - })? { - Ok(true) - } else { - Ok(false) - } +pub(crate) async fn check_file(conn: &Connection, filename: &Path) -> Result<bool> { + Ok(conn + .query(CHECK_FILE, params!(filename.to_str().unwrap())) + .await? + .next() + .await? + .unwrap() + .get::<bool>(0)?) } -pub(crate) fn init_clean_files(conn: &Connection) -> Result<Vec<PathBuf>, rusqlite::Error> { - conn.execute(DEDUPE_DB, ())?; - let mut stmt = conn.prepare(FETCH_FILES)?; - let mut rows = stmt.query(())?; +pub(crate) async fn init_clean_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> { + conn.execute(DEDUPE_DB, ()).await?; + let mut rows = conn.query(FETCH_FILES, ()).await?; let mut files = Vec::new(); - while let Ok(Some(row)) = rows.next() { + while let Ok(Some(row)) = rows.next().await { let path: String = row.get(0)?; files.push(PathBuf::from(path)); } + Ok(files) } -pub(crate) fn remove_file(conn: &Connection, filename: &Path) -> Result<()> { - conn.execute(REMOVE_FILE, params!(filename.to_str().unwrap()))?; +pub(crate) async fn remove_file(conn: &Connection, filename: &Path) -> Result<()> { + conn.execute(REMOVE_FILE, params!(filename.to_str().unwrap())) + .await?; Ok(()) } -pub(crate) fn get_toencode_files(conn: &Connection) -> Result<Vec<PathBuf>, rusqlite::Error> { - let mut stmt = conn.prepare(TOENCODE_PATHS)?; - let mut rows = stmt.query(())?; +pub(crate) async fn get_toencode_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> { + let mut rows = conn.query(TOENCODE_PATHS, ()).await?; let mut files: Vec<PathBuf> = Vec::new(); - while let Ok(Some(row)) = rows.next() { + while let Ok(Some(row)) = rows.next().await { let path: String = row.get(0)?; files.push(PathBuf::from(path)); } + Ok(files) } -pub(crate) fn get_toencode_number(conn: &Connection) -> Result<u64, rusqlite::Error> { - conn.query_one(TOENCODE_NUMBER, (), |row| { - let num: u64 = row.get(0)?; - Ok(num) - }) +pub(crate) async fn get_toencode_number(conn: &Connection) -> Result<u64, turso::Error> { + conn.query(TOENCODE_NUMBER, ()) + .await? + .next() + .await? + .unwrap() + .get::<u64>(0) } -pub(crate) fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> { - Ok( - conn.query_one(GET_MODTIME, params![file.to_str().unwrap()], |row| { - let modtime: u64 = row.get(0)?; - Ok(modtime) - })?, - ) +pub(crate) async fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> { + Ok(conn + .query(GET_MODTIME, params![file.to_str().unwrap()]) + .await? + .next() + .await? + .unwrap() + .get::<u64>(0)?) } -pub(crate) fn vacuum(conn: &Connection) -> Result<()> { - conn.execute("VACUUM", ())?; +pub(crate) async fn vacuum(conn: &Connection) -> Result<()> { + conn.execute("VACUUM", ()).await?; Ok(()) } @@ -129,9 +138,11 @@ pub(crate) fn vacuum(conn: &Connection) -> Result<()> { mod tests { use super::*; + use macro_rules_attribute::apply; + use smol_macros::{Executor, test}; - #[test] - fn check_localfiles() { + #[apply(test!)] + async fn check_localfiles(ex: &Executor<'_>) { let dbname = PathBuf::from("temp1.db"); let filenames = [ "./samples/16bit.flac", @@ -139,61 +150,69 @@ mod tests { "./samples/32bit.flac", ]; let mut counter = 0; - let conn = init_connection(Some(&dbname)).unwrap(); - for file in filenames { - let filename = PathBuf::from(file); - insert_file(&conn, &filename).unwrap(); - } - let mut stmt = conn.prepare(TOENCODE_PATHS).unwrap(); - let mut returned = stmt.query(()).unwrap(); - - while let Ok(Some(_)) = returned.next() { - counter += 1 - } - std::fs::remove_file(dbname).unwrap(); - assert!(counter == 0) + ex.spawn(async move { + let db = init_db(Some(&dbname)).await.unwrap(); + let conn = db.connect().unwrap(); + for file in filenames { + let path = PathBuf::from(file); + insert_file(&conn, &path).await.unwrap(); + } + let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap(); + while let Ok(Some(_)) = returned.next().await { + counter += 1 + } + std::fs::remove_file(dbname).unwrap(); + assert!(counter == 0) + }) + .await; } - #[test] - fn check_update() { + #[apply(test!)] + async fn check_update(ex: &Executor<'_>) { let dbname = PathBuf::from("temp2.db"); let filenames = [ "./samples/16bit.flac", "./samples/24bit.flac", "./samples/32bit.flac", ]; - let conn = init_connection(Some(&dbname)).unwrap(); - for file in filenames { - insert_file(&conn, &Path::new(file).canonicalize().unwrap()).unwrap(); - } - - conn.execute( - UPDATE_ITEM, - params![ - Path::new("./samples/16bit.flac") - .canonicalize() - .unwrap() - .to_str() - .unwrap(), - true, - "" - ], - ) - .unwrap(); + ex.spawn(async move { + let db = init_db(Some(&dbname)).await.unwrap(); + let conn = db.connect().unwrap(); + for file in filenames { + insert_file(&conn, &Path::new(file).canonicalize().unwrap()) + .await + .unwrap(); + } + conn.execute( + UPDATE_FILE, + params![ + Path::new("./samples/16bit.flac") + .canonicalize() + .unwrap() + .to_str() + .unwrap(), + true, + "" + ], + ) + .await + .unwrap(); - update_file( - &conn, - &Path::new("./samples/16bit.flac").canonicalize().unwrap(), - ) - .unwrap(); + update_file( + &conn, + &Path::new("./samples/16bit.flac").canonicalize().unwrap(), + ) + .await + .unwrap(); - let mut stmt = conn.prepare(TOENCODE_PATHS).unwrap(); - let mut returned = stmt.query(()).unwrap(); - let mut counter = 0; - while let Ok(Some(_)) = returned.next() { - counter += 1 - } - std::fs::remove_file(dbname).unwrap(); - assert!(counter == 0) + let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap(); + let mut counter = 0; + while let Ok(Some(_)) = returned.next().await { + counter += 1 + } + std::fs::remove_file(dbname).unwrap(); + assert!(counter == 0) + }) + .await; } } diff --git a/src/files.rs b/src/files.rs index 63c87fa..e3a31ca 100644 --- a/src/files.rs +++ b/src/files.rs @@ -3,19 +3,19 @@ use crate::flac::handle_encode; use anyhow::{Result, anyhow}; #[cfg(not(test))] use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle}; -use rusqlite::Connection; use std::{ error::Error, fmt::Display, path::{Path, PathBuf}, sync::{ - Arc, Mutex, + Arc, atomic::{AtomicBool, AtomicUsize, Ordering}, mpsc, }, thread::{self, sleep}, time::{Duration, UNIX_EPOCH}, }; +use turso::Connection; use walkdir::WalkDir; #[cfg(not(test))] @@ -51,26 +51,26 @@ impl Display for FileError { impl Error for FileError {} -fn handle_file(file: &Path, conn: &Connection) -> Result<()> { - if db::check_file(conn, file)? { +async fn handle_file(file: &Path, conn: &Connection) -> Result<()> { + if db::check_file(conn, file).await? { let modtime = file .metadata()? .modified()? .duration_since(UNIX_EPOCH)? .as_secs(); - let db_modtime = db::get_modtime(conn, file)?; + let db_modtime = db::get_modtime(conn, file).await?; if modtime != db_modtime { - db::update_file(conn, file)?; + db::update_file(conn, file).await?; } return Ok(()); } - db::insert_file(conn, file)?; + db::insert_file(conn, file).await?; Ok(()) } -pub(crate) fn index_files_recursively( +pub(crate) async fn index_files_recursively( path: &Path, conn: &Connection, handler: Arc<AtomicBool>, @@ -92,6 +92,8 @@ pub(crate) fn index_files_recursively( let newhandler = handler.clone(); + let newhandler = handler.clone(); + #[allow(unused_variables)] s.spawn(move || { for entry in WalkDir::new(&abspath) { @@ -108,8 +110,6 @@ pub(crate) fn index_files_recursively( let _ = filesend.send(path.to_owned()); #[cfg(not(test))] newbar.inc_length(1); - } - } } else { break; } @@ -120,7 +120,7 @@ pub(crate) fn index_files_recursively( && handler.load(Ordering::SeqCst) { #[allow(unused_variables)] - if let Err(error) = handle_file(&path, conn) { + if let Err(error) = smol::block_on( async {handle_file(&path, conn)}).await { #[cfg(not(test))] bar.println(format!("{}", FileError::new(&path, error))); } else { @@ -141,22 +141,20 @@ pub(crate) fn index_files_recursively( Ok(()) } -pub(crate) fn reencode_files( - conn: Connection, +pub async fn reencode_files( + conn: &Connection, handler: Arc<AtomicBool>, threads: usize, ) -> Result<()> { #[cfg(not(test))] let bar = ProgressBar::with_draw_target( - Some(db::get_toencode_number(&conn)?), + Some(db::get_toencode_number(conn).await?), ProgressDrawTarget::stdout_with_hz(60), ) .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-")) .with_message("Reencoding"); - let mut files = db::get_toencode_files(&conn)?.into_iter(); - - let lock = Arc::new(Mutex::new(conn)); + let mut files = db::get_toencode_files(conn).await?.into_iter(); let thread_counter = Arc::new(AtomicUsize::new(0)); @@ -177,17 +175,19 @@ pub(crate) fn reencode_files( thread_counter.fetch_add(1, Ordering::Relaxed); let handler = handler.clone(); - let lock = lock.clone(); #[cfg(not(test))] let bar = bar.clone(); let thread_counter = thread_counter.clone(); + let conn = conn.clone(); s.spawn(move || { match handle_encode(&file, handler) { Err(error) => eprintln!("{}", FileError::new(&file, error)), Ok(false) => { - if let Err(error) = db::update_file(&lock.lock().unwrap(), &file) { - eprintln!("{}", FileError::new(&file, error)); + if let Err(error) = + smol::block_on(async { db::update_file(&conn, &file).await }) + { + eprintln!("{}", FileError::new(&file, error)) } #[cfg(not(test))] bar.inc(1) @@ -210,8 +210,8 @@ pub(crate) fn reencode_files( Ok(()) } -pub(crate) fn clean_files(conn: &Connection, handler: Arc<AtomicBool>) -> Result<()> { - let files = db::init_clean_files(conn)?; +pub async fn clean_files(conn: &Connection, handler: Arc<AtomicBool>) -> Result<()> { + let files = db::init_clean_files(conn).await?; #[cfg(not(test))] let spinner = ProgressBar::with_draw_target(None, ProgressDrawTarget::stdout_with_hz(60)) @@ -222,17 +222,20 @@ pub(crate) fn clean_files(conn: &Connection, handler: Arc<AtomicBool>) -> Result files.iter().for_each(|file| { #[allow(clippy::collapsible_if)] if handler.load(Ordering::SeqCst) && !file.exists() { - if let Err(error) = db::remove_file(conn, file) { + #[cfg(not(test))] + let spinner = spinner.clone(); + if let Err(error) = smol::block_on(async { db::remove_file(conn, file).await }) { eprintln!("{}", FileError::new(file, error)) }; #[cfg(not(test))] spinner.inc(1); } }); + #[cfg(not(test))] spinner.finish(); - db::vacuum(conn)?; + db::vacuum(conn).await?; Ok(()) } @@ -240,52 +243,65 @@ pub(crate) fn clean_files(conn: &Connection, handler: Arc<AtomicBool>) -> Result #[cfg(test)] mod tests { use super::*; + use macro_rules_attribute::apply; + use smol_macros::{Executor, test}; - #[test] - fn test_index_lots_of_files() { + #[apply(test!)] + async fn test_index_lots_of_files(ex: &Executor<'_>) { let dbname = PathBuf::from("temp3.db"); let handler = Arc::new(AtomicBool::new(true)); - let conn = db::init_connection(Some(&dbname)).unwrap(); - index_files_recursively(Path::new("./testfiles"), &conn, handler).unwrap(); - std::fs::remove_file(dbname).unwrap(); + ex.spawn(async { + let db = db::init_db(Some(&dbname)).await.unwrap(); + let conn = db.connect().unwrap(); + index_files_recursively(Path::new("./testfiles"), &conn, handler).unwrap(); + std::fs::remove_file(dbname).unwrap(); + }) + .await } - #[test] - fn test_clean_files() { + #[apply(test!)] + async fn test_clean_files(ex: &Executor<'_>) { let dbname = PathBuf::from("temp4.db"); let handler = Arc::new(AtomicBool::new(true)); - let conn = db::init_connection(Some(&dbname)).unwrap(); - let filenames = [ - "./samples/16bit.flac", - "./samples/24bit.flac", - "./samples/32bit.flac", - "./samples/nonexisting.flac", - ]; - std::fs::copy("./samples/32bit.flac", "./samples/nonexisting.flac").unwrap(); - for file in filenames { - let filename = PathBuf::from(file); - db::insert_file(&conn, &filename).unwrap(); - } + ex.spawn(async { + let db = db::init_db(Some(&dbname)).await.unwrap(); + let conn = db.connect().unwrap(); + let filenames = [ + "./samples/16bit.flac", + "./samples/24bit.flac", + "./samples/32bit.flac", + "./samples/nonexisting.flac", + ]; + std::fs::copy("./samples/32bit.flac", "./samples/nonexisting.flac").unwrap(); + for file in filenames { + let filename = PathBuf::from(file); + db::insert_file(&conn, &filename).await.unwrap(); + } - std::fs::remove_file("./samples/nonexisting.flac").unwrap(); + std::fs::remove_file("./samples/nonexisting.flac").unwrap(); - clean_files(&conn, handler).unwrap(); - let counter = db::init_clean_files(&conn).unwrap().len(); - std::fs::remove_file(dbname).unwrap(); - assert!(counter == 3) + clean_files(&conn, handler).await.unwrap(); + let counter = db::init_clean_files(&conn).await.unwrap().len(); + std::fs::remove_file(dbname).unwrap(); + assert!(counter == 3) + }) + .await; } - #[test] - fn test_reencode_lots_of_files() { + #[apply(test!)] + async fn test_reencode_lots_of_files(ex: &Executor<'_>) { let dbname = PathBuf::from("temp5.db"); let handler = Arc::new(AtomicBool::new(true)); - let conn = db::init_connection(Some(&dbname)).unwrap(); - let temp = handler.clone(); - index_files_recursively(Path::new("./testfiles"), &conn, temp).unwrap(); - println!("\n{}", db::get_toencode_number(&conn).unwrap()); - reencode_files(conn, handler, 4).unwrap(); - let conn = db::init_connection(Some(&dbname)).unwrap(); - println!("\n{}", db::get_toencode_number(&conn).unwrap()); - std::fs::remove_file(dbname).unwrap(); + ex.spawn(async { + let db = db::init_db(Some(&dbname)).await.unwrap(); + let conn = db.connect().unwrap(); + let temp = handler.clone(); + index_files_recursively(Path::new("./testfiles"), &conn, temp).unwrap(); + println!("\n{}", db::get_toencode_number(&conn).await.unwrap()); + reencode_files(&conn, handler, 4).await.unwrap(); + println!("\n{}", db::get_toencode_number(&conn).await.unwrap()); + std::fs::remove_file(dbname).unwrap(); + }) + .await; } } diff --git a/src/main.rs b/src/main.rs index de3da62..e9543f8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,6 @@ -mod db; -mod files; -mod flac; +pub(crate) mod db; +pub(crate) mod files; +pub(crate) mod flac; use anyhow::Result; use clap::{Arg, ArgAction, Command, ValueHint, command, value_parser}; use clap_complete::{Generator, Shell, generate}; @@ -90,30 +90,32 @@ fn main() -> Result<()> { r.store(false, Ordering::SeqCst); })?; - let conn = db::init_connection(args.get_one::<PathBuf>("db"))?; + smol::block_on(async move { + let path = args.get_one::<PathBuf>("db"); + let db = db::init_db(path).await?; - let path = args.get_one::<PathBuf>("path"); + if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") { + let count = db::get_toencode_number(&db.connect()?).await?; + println!("Files to reencode:\t{}", style(count).green()); + return Ok(()); + } - if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") { - let count = db::get_toencode_number(&conn)?; - println!("Files to reencode:\t{}", style(count).green()); - return Ok(()); - } + if let Some(realpath) = path { + let hanlder = running.clone(); + files::index_files_recursively(realpath, &db.connect()?, hanlder)?; + } - if let Some(realpath) = path { - let hanlder = running.clone(); - files::index_files_recursively(realpath, &conn, hanlder)?; - } + if args.get_flag("clean") { + let handler = running.clone(); + files::clean_files(&db.connect()?, handler).await?; + } - if args.get_flag("clean") { - let handler = running.clone(); - files::clean_files(&conn, handler)?; - } + if args.get_flag("doit") { + let hanlder = running.clone(); + let threads = *args.get_one::<usize>("threads").unwrap(); + files::reencode_files(&db.connect()?, hanlder, threads).await?; + } - if args.get_flag("doit") { - let hanlder = running.clone(); - let threads = *args.get_one::<usize>("threads").unwrap(); - files::reencode_files(conn, hanlder, threads)?; - } - Ok::<(), anyhow::Error>(()) + Ok::<(), anyhow::Error>(()) + }) } |