diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/db.rs | 280 | ||||
| -rw-r--r-- | src/files.rs | 161 | ||||
| -rw-r--r-- | src/main.rs | 17 |
3 files changed, 223 insertions, 235 deletions
@@ -1,9 +1,11 @@ use anyhow::{Result, anyhow}; use directories::BaseDirs; -use libsql::{Builder, Connection, params}; +use r2d2::{Pool, PooledConnection}; +use r2d2_sqlite::SqliteConnectionManager; +use rusqlite::params; use std::{ path::{Path, PathBuf}, - time::{Duration, UNIX_EPOCH}, + time::UNIX_EPOCH, }; use crate::flac::{CURRENT_VENDOR, get_vendor}; @@ -14,27 +16,39 @@ const REPLACE_ITEM: &str = "REPLACE INTO flacs (path, toencode, modtime) VALUES const TOENCODE_QUERY: &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)"; -const FETCH_MODTIME: &str = "SELECT modtime FROM flacs WHERE path = ?1"; const FETCH_FILES: &str = "SELECT path FROM flacs"; const REMOVE_FILE: &str = "DELETE FROM flacs WHERE path = ?1"; 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"; -#[derive(Debug, Clone)] -pub struct Database(Connection); +pub fn open_db(path: Option<impl AsRef<Path>>) -> Result<Pool<SqliteConnectionManager>> { + if let Some(file) = path { + let manager = SqliteConnectionManager::file(file); + let pool = Pool::builder().build(manager)?; + let conn = pool.get()?; + conn.execute(TABLE_CREATE, ())?; + Ok(pool) + } else if let Some(base_dir) = BaseDirs::new() { + let file = Path::new(base_dir.data_dir()).join("reencoder.db"); + let manager = SqliteConnectionManager::file(file); + let pool = Pool::builder().build(manager)?; + let conn = pool.get()?; + conn.execute(TABLE_CREATE, ())?; + Ok(pool) + } else { + Err(anyhow!("Failed to locate data directory")) + } +} -impl Database { - pub async fn new(path: impl AsRef<Path>) -> Result<Self> { - let conn = Builder::new_local(path.as_ref().to_str().unwrap()) - .build() - .await? - .connect()?; - conn.execute(TABLE_CREATE, ()).await?; +pub struct Database(pub PooledConnection<SqliteConnectionManager>); - Ok(Database(conn)) +impl Database { + pub fn new(conn: PooledConnection<SqliteConnectionManager>) -> Self { + Database(conn) } - pub async fn insert_file(&self, filename: impl AsRef<Path>) -> Result<()> { + pub fn insert_file(&self, filename: impl AsRef<Path>) -> Result<()> { let toencode = !matches!(get_vendor(&filename)?.as_str(), CURRENT_VENDOR); let modtime = filename @@ -44,17 +58,15 @@ impl Database { .duration_since(UNIX_EPOCH)? .as_secs(); - self.0 - .execute( - ADD_NEW_ITEM, - params![filename.as_ref().to_str().unwrap(), toencode, modtime], - ) - .await?; + self.0.execute( + ADD_NEW_ITEM, + params![filename.as_ref().to_str().unwrap(), toencode, modtime], + )?; Ok(()) } - pub async fn update_file(&self, filename: impl AsRef<Path>) -> Result<()> { + pub fn update_file(&self, filename: impl AsRef<Path>) -> Result<()> { let modtime = filename .as_ref() .metadata()? @@ -62,176 +74,146 @@ impl Database { .duration_since(UNIX_EPOCH)? .as_secs(); - self.0 - .execute( - REPLACE_ITEM, - params![filename.as_ref().to_str().unwrap(), false, modtime], - ) - .await?; + self.0.execute( + REPLACE_ITEM, + params![filename.as_ref().to_str().unwrap(), false, modtime], + )?; Ok(()) } - pub async fn check_file(&self, filename: impl AsRef<Path>) -> Result<bool> { - if let Some(row) = self - .0 - .query(CHECK_FILE, params!(filename.as_ref().to_str().unwrap())) - .await? - .next() - .await? - { - Ok(matches!(row.get_value(0)?, libsql::Value::Integer(1))) + pub fn check_file(&self, filename: impl AsRef<Path>) -> Result<bool> { + if self.0.query_one( + CHECK_FILE, + params!(filename.as_ref().to_str().unwrap()), + |row| { + let num: bool = row.get(0)?; + Ok(num) + }, + )? { + Ok(true) } else { - Err(anyhow!("database error")) + Ok(false) } } - pub async fn get_modtime(&self, filename: impl AsRef<Path>) -> Result<u64> { - if let Some(row) = self - .0 - .query(FETCH_MODTIME, params!(filename.as_ref().to_str().unwrap())) - .await? - .next() - .await? - { - if let Some(sec) = row.get_value(0)?.as_integer() { - Ok(Duration::from_secs(*sec as u64).as_secs()) - } else { - Ok(Duration::from_secs(0).as_secs()) - } - } else { - Err(anyhow!("database error")) - } - } - - pub async fn init_clean_files(&self) -> Result<Vec<PathBuf>, libsql::Error> { - self.0.execute(DEDUPE_DB, ()).await?; - let mut rows = self.0.query(FETCH_FILES, ()).await?; + pub fn init_clean_files(&self) -> Result<Vec<PathBuf>, rusqlite::Error> { + self.0.execute(DEDUPE_DB, ())?; + let mut stmt = self.0.prepare(FETCH_FILES)?; + let mut rows = stmt.query(())?; let mut files = Vec::new(); - while let Ok(Some(row)) = rows.next().await { - files.push(PathBuf::from(row.get_value(0)?.as_text().unwrap())) + while let Ok(Some(row)) = rows.next() { + let path: String = row.get(0)?; + files.push(PathBuf::from(path)); } Ok(files) } - pub async fn remove_file(&self, filename: impl AsRef<Path>) -> Result<()> { + pub fn remove_file(&self, filename: impl AsRef<Path>) -> Result<()> { self.0 - .execute(REMOVE_FILE, params!(filename.as_ref().to_str().unwrap())) - .await?; + .execute(REMOVE_FILE, params!(filename.as_ref().to_str().unwrap()))?; Ok(()) } - pub async fn get_toencode_files(&self) -> Result<Vec<PathBuf>, libsql::Error> { - let mut rows = self.0.query(TOENCODE_QUERY, ()).await?; - let mut files = Vec::new(); - while let Ok(Some(row)) = rows.next().await { - files.push(PathBuf::from(row.get_value(0)?.as_text().unwrap())) + pub fn get_toencode_files(&self) -> Result<Vec<PathBuf>, rusqlite::Error> { + let mut stmt = self.0.prepare(TOENCODE_QUERY)?; + let mut rows = stmt.query(())?; + let mut files: Vec<PathBuf> = Vec::new(); + while let Ok(Some(row)) = rows.next() { + let path: String = row.get(0)?; + files.push(PathBuf::from(path)); } Ok(files) } - pub async fn get_toencode_number(&self) -> Result<i64> { - Ok(*self - .0 - .query(TOENCODE_NUMBER, ()) - .await? - .next() - .await? - .unwrap() - .get_value(0)? - .as_integer() - .unwrap()) + pub fn get_toencode_number(&self) -> Result<u64, rusqlite::Error> { + self.0.query_one(TOENCODE_NUMBER, (), |row| { + let num: u64 = row.get(0)?; + Ok(num) + }) } - pub async fn vaccum(&self) -> Result<()> { - self.0.execute("VACUUM", ()).await?; - Ok(()) + pub fn get_modtime(&self, file: impl AsRef<Path>) -> Result<u64> { + Ok(self.0.query_one( + GET_MODTIME, + params![file.as_ref().to_str().unwrap()], + |row| { + let modtime: u64 = row.get(0)?; + Ok(modtime) + }, + )?) } -} -pub async fn open_default_db() -> Result<Database> { - if let Some(base_dir) = BaseDirs::new() { - let db_name = Path::new(base_dir.data_dir()).join("reencoder.db"); - Ok(Database::new(db_name).await?) - } else { - Err(anyhow!("Failed to locate data directory")) + pub fn vaccum(&self) -> Result<()> { + self.0.execute("VACUUM", ())?; + Ok(()) } } #[cfg(test)] mod tests { - use macro_rules_attribute::apply; - use smol_macros::{Executor, test}; use super::*; - #[apply(test!)] - async fn check_localfiles(ex: &Executor<'_>) { - ex.spawn(async { - let dbname = String::from("temp1.db"); - let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"]; - let mut counter = 0; - let conn = Database::new(&dbname).await.unwrap(); - for file in filenames { - let _ = conn.insert_file(&file.to_string()).await; - } - let mut returned = conn.0.query(TOENCODE_QUERY, ()).await.unwrap(); + #[test] + fn check_localfiles() { + let dbname = String::from("temp1.db"); + let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"]; + let mut counter = 0; + let pool = open_db(Some(&dbname)).unwrap(); + let conn = Database::new(pool.get().unwrap()); + for file in filenames { + let _ = conn.insert_file(&file.to_string()); + } + let mut stmt = conn.0.prepare(TOENCODE_QUERY).unwrap(); + let mut returned = stmt.query(()).unwrap(); - while let Ok(Some(_)) = returned.next().await { - counter += 1 - } - std::fs::remove_file(dbname).unwrap(); - assert!(counter == 0) - }) - .await; + while let Ok(Some(_)) = returned.next() { + counter += 1 + } + std::fs::remove_file(dbname).unwrap(); + assert!(counter == 0) } - #[apply(test!)] - async fn check_update(ex: &Executor<'_>) { - ex.spawn(async { - let dbname = String::from("temp2.db"); - let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"]; - let conn = Database::new(&dbname).await.unwrap(); - for file in filenames { - let _ = conn - .insert_file(Path::new(file).canonicalize().unwrap()) - .await; - } - - let _ = conn - .0 - .execute( - REPLACE_ITEM, - params![ - Path::new("16bit.flac") - .canonicalize() - .unwrap() - .to_str() - .unwrap(), - true, - "" - ], - ) - .await; + #[test] + fn check_update() { + let dbname = String::from("temp2.db"); + let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"]; + let pool = open_db(Some(&dbname)).unwrap(); + let conn = Database::new(pool.get().unwrap()); + for file in filenames { + let _ = conn.insert_file(Path::new(file).canonicalize().unwrap()); + } - conn.update_file( + let _ = conn.0.execute( + REPLACE_ITEM, + params![ Path::new("16bit.flac") .canonicalize() .unwrap() .to_str() .unwrap(), - ) - .await - .unwrap(); + true, + "" + ], + ); - let mut returned = conn.0.query(TOENCODE_QUERY, ()).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; + conn.update_file( + Path::new("16bit.flac") + .canonicalize() + .unwrap() + .to_str() + .unwrap(), + ) + .unwrap(); + + let mut stmt = conn.0.prepare(TOENCODE_QUERY).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) } } diff --git a/src/files.rs b/src/files.rs index 797cad7..a92a07a 100644 --- a/src/files.rs +++ b/src/files.rs @@ -1,8 +1,9 @@ use anyhow::{Result, anyhow}; #[cfg(not(test))] use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle}; +use r2d2::Pool; +use r2d2_sqlite::SqliteConnectionManager; use rayon::prelude::*; -use smol::{Executor, fs::metadata}; use std::{ error::Error, fmt::Display, @@ -50,27 +51,24 @@ impl Display for FileError { impl Error for FileError {} -async fn handle_file(file: impl AsRef<Path>, conn: Database) -> Result<()> { - match conn.check_file(&file).await { - Ok(true) => { - let modtime = metadata(file.as_ref()) - .await? - .modified()? - .duration_since(UNIX_EPOCH)? - .as_secs(); - let db_time = conn.get_modtime(&file).await?; - if modtime != db_time { - if let Err(error) = conn.update_file(&file).await { - return Err(FileError::new(file, error).into()); - }; - } - return Ok(()); +fn handle_file(file: impl AsRef<Path>, conn: &Database) -> Result<()> { + if conn.check_file(&file)? { + let modtime = file + .as_ref() + .metadata()? + .modified()? + .duration_since(UNIX_EPOCH)? + .as_secs(); + let db_modtime = conn.get_modtime(&file)?; + if modtime != db_modtime { + if let Err(error) = conn.update_file(&file) { + return Err(FileError::new(&file, error).into()); + }; } - Err(error) => return Err(FileError::new(file, error).into()), - _ => {} + return Ok(()); } - if let Err(error) = conn.insert_file(&file).await { + if let Err(error) = conn.insert_file(&file) { return Err(FileError::new(file, error).into()); } @@ -79,7 +77,7 @@ async fn handle_file(file: impl AsRef<Path>, conn: Database) -> Result<()> { pub fn index_files_recursively( path: impl AsRef<Path>, - conn: &Database, + pool: &Pool<SqliteConnectionManager>, handler: Arc<AtomicBool>, ) -> Result<()> { if !path.as_ref().is_dir() { @@ -92,9 +90,7 @@ pub fn index_files_recursively( .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-")) .with_message("Indexing"); - let ex = Executor::new(); - - let mut tasks = Vec::new(); + let mut paths = Vec::new(); for entry in WalkDir::new(abspath) { if handler.load(Ordering::SeqCst) { @@ -103,25 +99,7 @@ pub fn index_files_recursively( continue; } if path.extension().is_some_and(|x| x == "flac") { - let newconn = conn.clone(); - let newhandler = handler.clone(); - #[cfg(not(test))] - let newbar = bar.clone(); - - tasks.push(ex.spawn(async move { - if newhandler.load(Ordering::SeqCst) { - match handle_file(&path, newconn).await { - Err(error) => Err(FileError::new(path, error)), - Ok(_) => { - #[cfg(not(test))] - newbar.inc(1); - Ok(()) - } - } - } else { - Ok(()) - } - })); + paths.push(path); #[cfg(not(test))] bar.inc_length(1); @@ -131,10 +109,21 @@ pub fn index_files_recursively( } } - tasks.par_iter_mut().for_each(|task| { + paths.par_iter().for_each(|file| { if handler.load(Ordering::SeqCst) { - if let Err(error) = smol::block_on(async { ex.run(task).await }) { - eprintln!("{error}") + let conn = match pool.get() { + Ok(conn) => Database::new(conn), + Err(error) => { + eprintln!("{}", FileError::new(file, error.into())); + return; + } + }; + + if let Err(error) = handle_file(file, &conn) { + eprintln!("{}", FileError::new(file, error)) + } else { + #[cfg(not(test))] + bar.inc(1); } } }); @@ -150,23 +139,35 @@ pub fn index_files_recursively( Ok(()) } -pub fn reencode_files(conn: &Database, handler: Arc<AtomicBool>) -> Result<()> { +pub fn reencode_files( + pool: &Pool<SqliteConnectionManager>, + handler: Arc<AtomicBool>, +) -> Result<()> { + let conn = Database::new(pool.get()?); #[cfg(not(test))] let bar = ProgressBar::with_draw_target( - Some(smol::block_on(async { conn.get_toencode_number().await })?.try_into()?), + Some(conn.get_toencode_number()?), ProgressDrawTarget::stdout_with_hz(60), ) .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-")) .with_message("Reencoding"); - let files = smol::block_on(async { conn.get_toencode_files().await })?; + let files = conn.get_toencode_files()?; + drop(conn); files.par_iter().for_each(|file| { if handler.load(Ordering::SeqCst) { + let conn = match pool.get() { + Ok(conn) => Database::new(conn), + Err(error) => { + eprintln!("{}", FileError::new(file, error.into())); + return; + } + }; if let Err(error) = handle_encode(file) { eprintln!("{}", FileError::new(file, error)); } else { - if let Err(error) = smol::block_on(async { conn.update_file(file).await }) { + if let Err(error) = conn.update_file(file) { eprintln!("{}", FileError::new(file, error)); } #[cfg(not(test))] @@ -186,8 +187,10 @@ pub fn reencode_files(conn: &Database, handler: Arc<AtomicBool>) -> Result<()> { Ok(()) } -pub fn clean_files(conn: &Database, handler: Arc<AtomicBool>) -> Result<()> { - let files = smol::block_on(async { conn.init_clean_files().await })?; +pub fn clean_files(pool: &Pool<SqliteConnectionManager>, handler: Arc<AtomicBool>) -> Result<()> { + let conn = Database::new(pool.get()?); + let files = conn.init_clean_files()?; + drop(conn); #[cfg(not(test))] let spinner = ProgressBar::with_draw_target(None, ProgressDrawTarget::stdout_with_hz(60)) @@ -195,7 +198,14 @@ pub fn clean_files(conn: &Database, handler: Arc<AtomicBool>) -> Result<()> { files.par_iter().for_each(|file| { if handler.load(Ordering::SeqCst) && !file.exists() { - if let Err(error) = smol::block_on(async { conn.remove_file(file).await }) { + let conn = match pool.get() { + Ok(conn) => Database::new(conn), + Err(error) => { + eprintln!("{}", FileError::new(file, error.into())); + return; + } + }; + if let Err(error) = conn.remove_file(file) { eprintln!("{}", FileError::new(file, error)) }; #[cfg(not(test))] @@ -205,7 +215,8 @@ pub fn clean_files(conn: &Database, handler: Arc<AtomicBool>) -> Result<()> { #[cfg(not(test))] spinner.finish(); - smol::block_on(async { conn.vaccum().await })?; + let conn = Database::new(pool.get()?); + conn.vaccum()?; Ok(()) } @@ -213,32 +224,28 @@ pub fn clean_files(conn: &Database, handler: Arc<AtomicBool>) -> Result<()> { #[cfg(test)] mod tests { use super::*; - use macro_rules_attribute::apply; - use smol_macros::{Executor, test}; + use crate::db::*; - #[apply(test!)] - async fn test_index_lots_of_files(ex: &Executor<'_>) { - ex.spawn(async { - let handler = Arc::new(AtomicBool::new(true)); - let conn = Database::new("temp3.db").await.unwrap(); - index_files_recursively(Path::new("./testfiles"), &conn, handler).unwrap(); - std::fs::remove_file("temp3.db").unwrap(); - }) - .await; + #[test] + fn test_index_lots_of_files() { + let handler = Arc::new(AtomicBool::new(true)); + let pool = open_db(Some("temp3.db")).unwrap(); + index_files_recursively(Path::new("./testfiles"), &pool, handler).unwrap(); + std::fs::remove_file("temp3.db").unwrap(); } - #[apply(test!)] - async fn test_reencode_lots_of_files(ex: &Executor<'_>) { - ex.spawn(async { - let handler = Arc::new(AtomicBool::new(true)); - let conn = Database::new("temp4.db").await.unwrap(); - let temp = handler.clone(); - index_files_recursively(Path::new("./testfiles"), &conn, temp).unwrap(); - println!("\n{}", conn.get_toencode_number().await.unwrap()); - reencode_files(&conn, handler).unwrap(); - println!("\n{}", conn.get_toencode_number().await.unwrap()); - std::fs::remove_file("temp4.db").unwrap(); - }) - .await; + #[test] + fn test_reencode_lots_of_files() { + let handler = Arc::new(AtomicBool::new(true)); + let pool = open_db(Some("temp4.db")).unwrap(); + let temp = handler.clone(); + index_files_recursively(Path::new("./testfiles"), &pool, temp).unwrap(); + let conn = Database::new(pool.get().unwrap()); + println!("\n{}", conn.get_toencode_number().unwrap()); + drop(conn); + reencode_files(&pool, handler).unwrap(); + let conn = Database::new(pool.get().unwrap()); + println!("\n{}", conn.get_toencode_number().unwrap()); + std::fs::remove_file("temp4.db").unwrap(); } } diff --git a/src/main.rs b/src/main.rs index 206a13b..da62d1e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,8 @@ use std::{ }, }; +use crate::db::Database; + fn build_cli() -> Command { command!() .arg( @@ -90,16 +92,13 @@ fn main() -> Result<()> { r.store(false, Ordering::SeqCst); })?; - let conn = if let Some(path) = args.get_one::<PathBuf>("db") { - smol::block_on(async { db::Database::new(path).await })? - } else { - smol::block_on(async { db::open_default_db().await })? - }; + let dbpool = db::open_db(args.get_one::<PathBuf>("db"))?; let path = args.get_one::<PathBuf>("path"); if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") { - let count = smol::block_on(async { conn.get_toencode_number().await })?; + let conn = Database::new(dbpool.get()?); + let count = conn.get_toencode_number()?; println!("Files to reencode:\t{}", style(count).green()); return Ok(()); } @@ -110,17 +109,17 @@ fn main() -> Result<()> { if let Some(realpath) = path { let hanlder = running.clone(); - pool.install(|| files::index_files_recursively(realpath, &conn, hanlder))?; + pool.install(|| files::index_files_recursively(realpath, &dbpool, hanlder))?; } if args.get_flag("clean") { let handler = running.clone(); - pool.install(|| files::clean_files(&conn, handler))?; + pool.install(|| files::clean_files(&dbpool, handler))?; } if args.get_flag("doit") { let hanlder = running.clone(); - pool.install(|| files::reencode_files(&conn, hanlder))?; + pool.install(|| files::reencode_files(&dbpool, hanlder))?; } Ok::<(), anyhow::Error>(()) } |