tyler/proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

src/backup/chunk_stream.rs: use more generics

Browse Source
This commit is contained in:
Dietmar Maurer 2019-05-22 09:18:05 +02:00
parent 169c0e060f
commit c052be5c86
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/backup/chunk_stream.rs
View File

@ -5,21 +5,24 @@ use futures::{Async, Poll};
use futures::stream::Stream; use futures::stream::Stream;
/// Split input stream into dynamic sized chunks /// Split input stream into dynamic sized chunks
pub struct ChunkStream<S: Stream<Item=Vec<u8>, Error=Error>> { pub struct ChunkStream<S> {
input: S, input: S,
chunker: Chunker, chunker: Chunker,
buffer: Option<Vec<u8>>, buffer: Option<Vec<u8>>,
scan: Option<Vec<u8>>, scan: Option<Vec<u8>>,
} }
impl <S: Stream<Item=Vec<u8>, Error=Error>> ChunkStream<S> { impl <S> ChunkStream<S> {
pub fn new(input: S) -> Self { pub fn new(input: S) -> Self {
Self { input, chunker: Chunker::new(4 * 1024 * 1024), buffer: None, scan: None} Self { input, chunker: Chunker::new(4 * 1024 * 1024), buffer: None, scan: None}
} }
} }
impl <S: Stream<Item=Vec<u8>, Error=Error>> Stream for ChunkStream<S> { impl <S> Stream for ChunkStream<S>
where S: Stream,
S::Item: AsRef<[u8]>,
S::Error: Into<Error>,
{
type Item = Vec<u8>; type Item = Vec<u8>;
type Error = Error; type Error = Error;
@ -49,7 +52,7 @@ impl <S: Stream<Item=Vec<u8>, Error=Error>> Stream for ChunkStream<S> {
match self.input.poll() { match self.input.poll() {
Err(err) => { Err(err) => {
return Err(err); return Err(err.into());
} }
Ok(Async::NotReady) => { Ok(Async::NotReady) => {
return Ok(Async::NotReady); return Ok(Async::NotReady);
@ -66,7 +69,7 @@ impl <S: Stream<Item=Vec<u8>, Error=Error>> Stream for ChunkStream<S> {
} }
Ok(Async::Ready(Some(data))) => { Ok(Async::Ready(Some(data))) => {
let scan = self.scan.get_or_insert_with(|| Vec::with_capacity(1024*1024)); let scan = self.scan.get_or_insert_with(|| Vec::with_capacity(1024*1024));
scan.extend(data); scan.extend(data.as_ref());
} }
} }
} }