[pbs-devel] applied-both: [PATCH proxmox 1/2] proxmox-compression: make ZstdEncoder stream a bit more generic

[Wolfgang Bumiller <w.bumiller@proxmox.com>]

applied both patches for proxmox-compression & bumped

On Tue, May 31, 2022 at 01:17:20PM +0200, Dominik Csapak wrote:
> by not requiring the 'anyhow::Error' type from the source, but only
> that it implements 'Into<Error>'. This way, we can also accept a stream
> that produces e.g. an io::Error
> 
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
>  proxmox-compression/src/zstd.rs | 15 ++++++++++-----
>  1 file changed, 10 insertions(+), 5 deletions(-)
> 
> diff --git a/proxmox-compression/src/zstd.rs b/proxmox-compression/src/zstd.rs
> index d9e5826..3d4b2ab 100644
> --- a/proxmox-compression/src/zstd.rs
> +++ b/proxmox-compression/src/zstd.rs
> @@ -32,10 +32,11 @@ pub struct ZstdEncoder<'a, T> {
>      state: EncoderState,
>  }
>  
> -impl<'a, T, O> ZstdEncoder<'a, T>
> +impl<'a, T, O, E> ZstdEncoder<'a, T>
>  where
> -    T: Stream<Item = Result<O, Error>> + Unpin,
> +    T: Stream<Item = Result<O, E>> + Unpin,
>      O: Into<Bytes>,
> +    E: Into<Error>,
>  {
>      /// Returns a new [ZstdEncoder] with default level 3
>      pub fn new(inner: T) -> Result<Self, io::Error> {
> @@ -79,10 +80,11 @@ impl<'a, T> ZstdEncoder<'a, T> {
>      }
>  }
>  
> -impl<'a, T, O> Stream for ZstdEncoder<'a, T>
> +impl<'a, T, O, E> Stream for ZstdEncoder<'a, T>
>  where
> -    T: Stream<Item = Result<O, Error>> + Unpin,
> +    T: Stream<Item = Result<O, E>> + Unpin,
>      O: Into<Bytes>,
> +    E: Into<Error>,
>  {
>      type Item = Result<Bytes, Error>;
>  
> @@ -93,7 +95,10 @@ where
>              match this.state {
>                  EncoderState::Reading => {
>                      if let Some(res) = ready!(Pin::new(&mut this.inner).poll_next(cx)) {
> -                        let buf = res?;
> +                        let buf = match res {
> +                            Ok(buf) => buf,
> +                            Err(err) => return Poll::Ready(Some(Err(err.into()))),
> +                        };
>                          this.input_buffer = buf.into();
>                          this.state = EncoderState::Writing;
>                      } else {
> -- 
> 2.30.2