Module async_compression::stream
source · [−]async-compression::stream docs for migrationstream only.Expand description
Types which operate over Stream<Item = io::Result<Bytes>> streams, both encoders and
decoders for various formats.
The Stream is treated as a single byte-stream to be compressed/decompressed, each item is a
chunk of data from this byte-stream. There is not guaranteed to be a one-to-one relationship
between chunks of data from the underlying stream and the resulting compressed/decompressed
stream, the encoders and decoders will buffer the incoming data and choose their own boundaries
at which to yield a new item.
Deprecation Migration
This feature and module was deprecated because it’s choosing one point in a large solution
space of “stream of byte chunks” to represent an IO data stream, and the conversion between
these solutions and standard IO data streams like futures::io::AsyncBufRead /
tokio::io::AsyncBufRead should be zero-cost.
use bytes_05::Bytes;
use futures::{stream::Stream, TryStreamExt};
use std::io::Result;
/// For code that looks like this, choose one of the options below to replace it
fn from(
input: impl Stream<Item = Result<bytes_05::Bytes>>,
) -> impl Stream<Item = Result<bytes_05::Bytes>> {
#[allow(deprecated)]
async_compression::stream::GzipEncoder::new(input)
}
/// Direct replacement with `tokio` v0.2 and `bytes` v0.5 using `tokio-util` v0.3
fn tokio_02_bytes_05(
input: impl Stream<Item = Result<bytes_05::Bytes>>,
) -> impl Stream<Item = Result<bytes_05::Bytes>> {
tokio_util_03::codec::FramedRead::new(
async_compression::tokio_02::bufread::GzipEncoder::new(
tokio_02::io::stream_reader(input),
),
tokio_util_03::codec::BytesCodec::new(),
).map_ok(|bytes| bytes.freeze())
}
/// Upgrade replacement with `tokio` v0.3 and `bytes` v0.5 using `tokio-util` v0.4
fn tokio_03_bytes_05(
input: impl Stream<Item = Result<bytes_05::Bytes>>,
) -> impl Stream<Item = Result<bytes_05::Bytes>> {
tokio_util_04::io::ReaderStream::new(
async_compression::tokio_03::bufread::GzipEncoder::new(
tokio_util_04::io::StreamReader::new(input),
),
)
}
/// Upgrade replacement with `tokio` v0.3 and `bytes` v0.6 using `tokio-util` v0.5
fn tokio_03_bytes_06(
input: impl Stream<Item = Result<bytes_06::Bytes>>,
) -> impl Stream<Item = Result<bytes_06::Bytes>> {
tokio_util_05::io::ReaderStream::new(
async_compression::tokio_03::bufread::GzipEncoder::new(
tokio_util_05::io::StreamReader::new(input),
),
)
}
/// Upgrade replacement with `tokio` v1.0 and `bytes` v1.0 using `tokio-util` v0.6
fn tokio_bytes(
input: impl Stream<Item = Result<bytes::Bytes>>,
) -> impl Stream<Item = Result<bytes::Bytes>> {
tokio_util_06::io::ReaderStream::new(
async_compression::tokio::bufread::GzipEncoder::new(
tokio_util_06::io::StreamReader::new(input),
),
)
}
/// What if you didn't want anything to do with `bytes`, but just a `Vec<u8>` instead?
fn futures_vec(
input: impl Stream<Item = Result<Vec<u8>>> + Unpin,
) -> impl Stream<Item = Result<Vec<u8>>> {
use futures::io::AsyncReadExt;
futures::stream::try_unfold(
async_compression::futures::bufread::GzipEncoder::new(input.into_async_read()),
|mut encoder| async move {
let mut chunk = vec![0; 8 * 1024];
let len = encoder.read(&mut chunk).await?;
if len == 0 {
Ok(None)
} else {
chunk.truncate(len);
Ok(Some((chunk, encoder)))
}
})
}