fmt

using ZstdEncoder from ..:tokio::bufread instead of ..:tokio::write

app_async/src/main.rs

@@ -158,20 +158,18 @@ async fn read_from_db(id: u32) -> Option<Question> {
         id as usize - 1
     };
 
-    println!("{}", &index); // DEBUG
-
     match reader.get(index).await {
         Ok(question) => Some(question),
-        Err(e) => {
+        Err(_) => None,
-            println!("{:#?}", e); // DEBUG
-            None
-        }
     }
 }
 async fn write_db() {
     let (tx, rx) = mpsc::unbounded_channel::<Question>();
-    let (r, _) = tokio::join!(tokio::spawn(zip_reader_task(tx)), db_writer_task(rx),);
+    tokio::try_join!(
-    r.expect("tokio join");
+        tokio::spawn(zip_reader_task(tx)),
+        tokio::spawn(db_writer_task(rx))
+    )
+    .expect("tokio join");
     println!("all done");
 }
 async fn db_writer_task(rx: UnboundedReceiver<Question>) {

lib/src/async_db.rs

@@ -1,7 +1,7 @@
 use std::{path::Path, sync::Arc};
 
 use async_compression::tokio::bufread::ZstdDecoder;
-use async_compression::tokio::write::ZstdEncoder;
+use async_compression::tokio::bufread::ZstdEncoder;
 use async_compression::Level;
 use futures::stream::StreamExt;
 use futures_core::stream::Stream;
@@ -78,9 +78,10 @@ where
 
         let item_data = bincode::encode_to_vec(item, BINCODE_CFG).str_err()?;
 
-        let mut zencoder = ZstdEncoder::with_quality(&mut self.data_buf, self.compress_lvl);
+        let mut zencoder = ZstdEncoder::with_quality(&item_data[..], self.compress_lvl);
-        zencoder.write_all(&item_data).await.str_err()?;
+        io::copy(&mut zencoder, &mut self.data_buf)
-        zencoder.flush().await.str_err()?;
+            .await
+            .str_err()?;
 
         self.table.push(pos);
 
@@ -116,7 +117,6 @@ where
             let pos_data = (pos + tab_size).to_le_bytes();
             self.out.write_all(&pos_data).await.str_err()?;
         }
-
         // copy data
         self.out.write_all(&self.data_buf[..]).await.str_err()?;