diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/db.rs | 51 | ||||
| -rw-r--r-- | src/files.rs | 58 | ||||
| -rw-r--r-- | src/main.rs | 9 |
3 files changed, 68 insertions, 50 deletions
@@ -1,11 +1,10 @@ use anyhow::{Result, anyhow}; use directories::BaseDirs; -use libsql::{Builder, Connection, params}; -use smol::stream::Stream; use std::{ path::Path, time::{Duration, UNIX_EPOCH}, }; +use turso::{Builder, Connection, Rows, params}; use crate::flac::{CURRENT_VENDOR, get_vendor}; @@ -26,7 +25,10 @@ pub struct Database(Connection); impl Database { pub async fn new(path: impl AsRef<Path>) -> Result<Self> { - let conn = Builder::new_local(path).build().await?.connect()?; + let conn = Builder::new_local(path.as_ref().to_str().unwrap()) + .build() + .await? + .connect()?; conn.execute(TABLE_CREATE, ()).await?; Ok(Database(conn)) @@ -78,7 +80,7 @@ impl Database { .next() .await? { - Ok(matches!(row.get_value(0)?, libsql::Value::Integer(1))) + Ok(matches!(row.get_value(0)?, turso::Value::Integer(1))) } else { Err(anyhow!("database error")) } @@ -102,11 +104,9 @@ impl Database { } } - pub async fn init_clean_files( - &self, - ) -> Result<impl Stream<Item = libsql::Result<libsql::Row>>> { + pub async fn init_clean_files(&self) -> Result<Rows, turso::Error> { self.0.execute(DEDUPE_DB, ()).await?; - Ok(self.0.query(FETCH_FILES, ()).await?.into_stream()) + self.0.query(FETCH_FILES, ()).await } pub async fn remove_file(&self, filename: impl AsRef<Path>) -> Result<()> { @@ -116,21 +116,21 @@ impl Database { Ok(()) } - pub async fn get_toencode_stream( - &self, - ) -> Result<impl Stream<Item = libsql::Result<libsql::Row>>> { - Ok(self.0.query(TOENCODE_QUERY, ()).await?.into_stream()) + pub async fn get_toencode_files(&self) -> Result<Rows, turso::Error> { + self.0.query(TOENCODE_QUERY, ()).await } - pub async fn get_toencode_number(&self) -> Result<u64> { - Ok(self + pub async fn get_toencode_number(&self) -> Result<i64> { + Ok(*self .0 .query(TOENCODE_NUMBER, ()) .await? .next() .await? .unwrap() - .get::<u64>(0)?) + .get_value(0)? + .as_integer() + .unwrap()) } pub async fn vaccum(&self) -> Result<()> { @@ -150,7 +150,6 @@ pub async fn open_default_db() -> Result<Database> { #[cfg(test)] mod tests { - use futures_util::StreamExt; use macro_rules_attribute::apply; use smol_macros::{Executor, test}; @@ -166,15 +165,9 @@ mod tests { for file in filenames { let _ = conn.insert_file(&file.to_string()).await; } - let returned = conn - .0 - .query(TOENCODE_QUERY, ()) - .await - .unwrap() - .into_stream(); - pin_utils::pin_mut!(returned); + let mut returned = conn.0.query(TOENCODE_QUERY, ()).await.unwrap(); - while let Some(Ok(_)) = returned.next().await { + while let Ok(Some(_)) = returned.next().await { counter += 1 } std::fs::remove_file(dbname).unwrap(); @@ -221,15 +214,9 @@ mod tests { .await .unwrap(); - let returned = conn - .0 - .query(TOENCODE_QUERY, ()) - .await - .unwrap() - .into_stream(); - pin_utils::pin_mut!(returned); + let mut returned = conn.0.query(TOENCODE_QUERY, ()).await.unwrap(); let mut counter = 0; - while let Some(Ok(_)) = returned.next().await { + while let Ok(Some(_)) = returned.next().await { counter += 1 } std::fs::remove_file(dbname).unwrap(); diff --git a/src/files.rs b/src/files.rs index 52789ce..13ba06a 100644 --- a/src/files.rs +++ b/src/files.rs @@ -134,39 +134,71 @@ pub fn index_files_recursively( #[cfg(not(test))] bar.inc_length(1); } + } else { + break; } } tasks.par_iter_mut().for_each(|task| { - if let Err(error) = smol::block_on(async { ex.run(task).await }) { - eprintln!("{error}") + if running.load(Ordering::SeqCst) { + if let Err(error) = smol::block_on(async { ex.run(task).await }) { + eprintln!("{error}") + } } }); #[cfg(not(test))] { - if !running.load(Ordering::SeqCst) { - bar.abandon_with_message("Indexing aborted"); - } else { + if running.load(Ordering::SeqCst) { bar.finish_with_message("Finished indexing"); + } else { + bar.abandon_with_message("Indexing aborted"); } } Ok(()) } -/* pub fn reencode_files(conn: &Database) -> Result<()> { - let stream = conn.get_toencode_stream().await?; - pin_mut!(stream); +async fn get_reencode_vec(conn: &Database) -> Result<Vec<PathBuf>> { + let mut files = Vec::new(); + let mut rows = conn.get_toencode_files().await?; + while let Ok(Some(row)) = rows.next().await { + files.push(PathBuf::from(row.get_value(0)?.as_text().unwrap())) + } + Ok(files) +} +pub fn reencode_files(conn: &Database, running: Arc<AtomicBool>) -> Result<()> { #[cfg(not(test))] let bar = ProgressBar::with_draw_target( - Some(conn.get_toencode_number().await?), + Some(smol::block_on(async { conn.get_toencode_number().await })?.try_into()?), ProgressDrawTarget::stdout_with_hz(60), ) .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-")) .with_message("Reencoding"); - while let Some(Ok(row)) = stream.next().await { + let files = smol::block_on(async { get_reencode_vec(conn).await })?; + + files.par_iter().for_each(|file| { + if running.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_secs(3)); + #[cfg(not(test))] + bar.inc(1); + } else { + eprintln!("{}", FileError::new(file, anyhow!("cancelled"))) + } + }); + + #[cfg(not(test))] + { + if running.load(Ordering::SeqCst) { + bar.finish_with_message("Finished reencoding"); + } else { + bar.abandon_with_message("Reencoding aborted"); + } + } + Ok(()) + + /* while let Some(Ok(row)) = stream.next().await { let filename = Path::new(row.get_str(0)?).canonicalize()?; if filename.exists() { let newconn = conn.clone(); @@ -197,12 +229,10 @@ pub fn index_files_recursively( } }); } - } - - Ok(()) + } */ } -pub fn clean_files(conn: &Database) -> Result<()> { +/* pub fn clean_files(conn: &Database) -> Result<()> { let ex = Executor::new(); let mut tasks: JoinSet<std::result::Result<(), anyhow::Error>> = JoinSet::new(); diff --git a/src/main.rs b/src/main.rs index 6db57a3..0e28b1f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -114,12 +114,13 @@ fn main() -> Result<()> { pool.install(|| files::index_files_recursively(realpath, &conn, hanlder))?; } - /* if args.get_flag("clean") { + /* if args.get_flag("clean") { pool.install(|| files::clean_files(&conn))?; - } + } */ if args.get_flag("doit") { - files::reencode_files(&conn)?; - } */ + let hanlder = running.clone(); + pool.install(|| files::reencode_files(&conn, hanlder))?; + } Ok::<(), anyhow::Error>(()) } |