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;
@@ -59,7 +59,7 @@ where
         let out = fs::File::create(path).await.str_err()?;
         let out = io::BufWriter::with_capacity(opts.out_buf_size, out);
         let data_buf: Vec<u8> = Vec::with_capacity(opts.data_buf_size);
-
+        
         let compress_lvl = opts.compress_lvl;
 
         let table: Vec<LSize> = vec![];
@@ -78,10 +78,9 @@ 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);
-        zencoder.write_all(&item_data).await.str_err()?;
-        zencoder.flush().await.str_err()?;
-
+        let mut zencoder = ZstdEncoder::with_quality(&item_data[..], self.compress_lvl);
+        io::copy(&mut zencoder, &mut self.data_buf).await.str_err()?;
+        
         self.table.push(pos);
 
         Ok(())
@@ -116,7 +115,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()?;