try brotli

Author: dignifiedquire

Cargo.lock

@@ -14,7 +14,7 @@ rust-version = "1.63"
 [dependencies]
 abao = { version = "0.2.0", features = ["group_size_16k", "tokio_io"], default-features = false }
 anyhow = { version = "1", features = ["backtrace"] }
-async-compression = { version = "0.3.15", features = ["tokio", "zstd"] }
+async-compression = { version = "0.3.15", features = ["tokio", "brotli"] }
 base64 = "0.21.0"
 blake3 = "1.3.3"
 bytes = "1"

Cargo.toml

@@ -106,7 +106,7 @@ impl Stats {
 pub struct DataStream(AsyncSliceDecoder<RecvStream>);
 
 type RecvStream =
-    async_compression::tokio::bufread::ZstdDecoder<tokio::io::BufReader<quinn::RecvStream>>;
+    async_compression::tokio::bufread::BrotliDecoder<tokio::io::BufReader<quinn::RecvStream>>;
 
 impl DataStream {
     fn new(inner: RecvStream, hash: Hash) -> Self {
@@ -284,7 +284,7 @@ async fn handle_blob_response(
                     assert!(buffer.is_empty());
                     // Decompress data
                     let decompress_reader =
-                        async_compression::tokio::bufread::ZstdDecoder::new(reader);
+                        async_compression::tokio::bufread::BrotliDecoder::new(reader);
                     let decoder = DataStream::new(decompress_reader, hash);
                     Ok(decoder)
                 }

src/get.rs

@@ -737,14 +737,10 @@ async fn send_blob<W: AsyncWrite + Unpin + Send + 'static>(
             // taking a reference does not work. spawn_blocking requires
             // 'static lifetime.
 
-            writer = tokio::task::spawn_blocking(move || {
-                // Compress data
-                let mut compressed_writer =
-                    async_compression::tokio::write::ZstdEncoder::with_quality(
-                        writer,
-                        async_compression::Level::Fastest,
-                    );
+            // Compress data
+            let mut compressed_writer = async_compression::tokio::write::BrotliEncoder::new(writer);
 
+            compressed_writer = tokio::task::spawn_blocking(move || {
                 let file_reader = std::fs::File::open(&path)?;
                 let outboard_reader = std::io::Cursor::new(outboard);
                 let mut wrapper = SyncIoBridge::new(&mut compressed_writer);
@@ -755,9 +751,12 @@ async fn send_blob<W: AsyncWrite + Unpin + Send + 'static>(
                     size,
                 );
                 let _copied = std::io::copy(&mut slice_extractor, &mut wrapper)?;
-                std::io::Result::Ok(compressed_writer.into_inner())
+                std::io::Result::Ok(compressed_writer)
             })
             .await??;
+            compressed_writer.flush().await?;
+            let writer = compressed_writer.into_inner();
+
             Ok((SentStatus::Sent, writer))
         }
         _ => {