app_async/src/main.rs

@@ -158,18 +158,20 @@ 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(_) => None,
+        Err(e) => {
+            println!("{:#?}", e); // DEBUG
+            None
+        }
     }
 }
 async fn write_db() {
     let (tx, rx) = mpsc::unbounded_channel::<Question>();
-    tokio::try_join!(
+    let (r, _) = tokio::join!(tokio::spawn(zip_reader_task(tx)), db_writer_task(rx),);
-        tokio::spawn(zip_reader_task(tx)),
+    r.expect("tokio join");
-        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::bufread::ZstdEncoder;
+use async_compression::tokio::write::ZstdEncoder;
 use async_compression::Level;
 use futures::stream::StreamExt;
 use futures_core::stream::Stream;
@@ -78,10 +78,9 @@ where
 
         let item_data = bincode::encode_to_vec(item, BINCODE_CFG).str_err()?;
 
-        let mut zencoder = ZstdEncoder::with_quality(&item_data[..], self.compress_lvl);
+        let mut zencoder = ZstdEncoder::with_quality(&mut self.data_buf, self.compress_lvl);
-        io::copy(&mut zencoder, &mut self.data_buf)
+        zencoder.write_all(&item_data).await.str_err()?;
-            .await
+        zencoder.flush().await.str_err()?;
-            .str_err()?;
 
         self.table.push(pos);
 
@@ -117,6 +116,7 @@ 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()?;