/rust/registry/src/index.crates.io-6f17d22bba15001f/flate2-1.0.30/src/zio.rs

Source (jump to first uncovered line)
use std::io;
use std::io::prelude::*;
use std::mem;

use crate::{Compress, Decompress, DecompressError, FlushCompress, FlushDecompress, Status};

#[derive(Debug)]
pub struct Writer<W: Write, D: Ops> {
    obj: Option<W>,
    pub data: D,
    buf: Vec<u8>,
}

pub trait Ops {
    type Flush: Flush;
    fn total_in(&self) -> u64;
    fn total_out(&self) -> u64;
    fn run(
        &mut self,
        input: &[u8],
        output: &mut [u8],
        flush: Self::Flush,
    ) -> Result<Status, DecompressError>;
    fn run_vec(
        &mut self,
        input: &[u8],
        output: &mut Vec<u8>,
        flush: Self::Flush,
    ) -> Result<Status, DecompressError>;
}

impl Ops for Compress {
    type Flush = FlushCompress;
    fn total_in(&self) -> u64 {
        self.total_in()
    }
    fn total_out(&self) -> u64 {
        self.total_out()
    }
    fn run(
        &mut self,
        input: &[u8],
        output: &mut [u8],
        flush: FlushCompress,
    ) -> Result<Status, DecompressError> {
        Ok(self.compress(input, output, flush).unwrap())
    }
    fn run_vec(
        &mut self,
        input: &[u8],
        output: &mut Vec<u8>,
        flush: FlushCompress,
    ) -> Result<Status, DecompressError> {
        Ok(self.compress_vec(input, output, flush).unwrap())
    }
}

impl Ops for Decompress {
    type Flush = FlushDecompress;
    fn total_in(&self) -> u64 {
        self.total_in()
    }
    fn total_out(&self) -> u64 {
        self.total_out()
    }
    fn run(
        &mut self,
        input: &[u8],
        output: &mut [u8],
        flush: FlushDecompress,
    ) -> Result<Status, DecompressError> {
        self.decompress(input, output, flush)
    }
    fn run_vec(
        &mut self,
        input: &[u8],
        output: &mut Vec<u8>,
        flush: FlushDecompress,
    ) -> Result<Status, DecompressError> {
        self.decompress_vec(input, output, flush)
    }
}

pub trait Flush {
    fn none() -> Self;
    fn sync() -> Self;
    fn finish() -> Self;
}

impl Flush for FlushCompress {
    fn none() -> Self {
        FlushCompress::None
    }

    fn sync() -> Self {
        FlushCompress::Sync
    }

    fn finish() -> Self {
        FlushCompress::Finish
    }
}

impl Flush for FlushDecompress {
    fn none() -> Self {
        FlushDecompress::None
    }

    fn sync() -> Self {
        FlushDecompress::Sync
    }

    fn finish() -> Self {
        FlushDecompress::Finish
    }
}

pub fn read<R, D>(obj: &mut R, data: &mut D, dst: &mut [u8]) -> io::Result<usize>
where
    R: BufRead,
    D: Ops,
{
    loop {
        let (read, consumed, ret, eof);
        {
            let input = obj.fill_buf()?;
            eof = input.is_empty();
            let before_out = data.total_out();
            let before_in = data.total_in();
            let flush = if eof {
                D::Flush::finish()
            } else {
                D::Flush::none()
            };
            ret = data.run(input, dst, flush);
            read = (data.total_out() - before_out) as usize;
            consumed = (data.total_in() - before_in) as usize;
        }
        obj.consume(consumed);

        match ret {
            // If we haven't ready any data and we haven't hit EOF yet,
            // then we need to keep asking for more data because if we
            // return that 0 bytes of data have been read then it will
            // be interpreted as EOF.
            Ok(Status::Ok | Status::BufError) if read == 0 && !eof && !dst.is_empty() => continue,
            Ok(Status::Ok | Status::BufError | Status::StreamEnd) => return Ok(read),

            Err(..) => {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "corrupt deflate stream",
                ))
            }
        }
    }
}

impl<W: Write, D: Ops> Writer<W, D> {
    pub fn new(w: W, d: D) -> Writer<W, D> {
        Writer {
            obj: Some(w),
            data: d,
            buf: Vec::with_capacity(32 * 1024),
        }
    }

    pub fn finish(&mut self) -> io::Result<()> {
        loop {
            self.dump()?;

            let before = self.data.total_out();
            self.data.run_vec(&[], &mut self.buf, D::Flush::finish())?;
            if before == self.data.total_out() {
                return Ok(());
            }
        }
    }

    pub fn replace(&mut self, w: W) -> W {
        self.buf.truncate(0);
        mem::replace(self.get_mut(), w)
    }

    pub fn get_ref(&self) -> &W {
        self.obj.as_ref().unwrap()
    }

    pub fn get_mut(&mut self) -> &mut W {
        self.obj.as_mut().unwrap()
    }

    // Note that this should only be called if the outer object is just about
    // to be consumed!
    //
    // (e.g. an implementation of `into_inner`)
    pub fn take_inner(&mut self) -> W {
        self.obj.take().unwrap()
    }

    pub fn is_present(&self) -> bool {
        self.obj.is_some()
    }

    // Returns total written bytes and status of underlying codec
    pub(crate) fn write_with_status(&mut self, buf: &[u8]) -> io::Result<(usize, Status)> {
        // miniz isn't guaranteed to actually write any of the buffer provided,
        // it may be in a flushing mode where it's just giving us data before
        // we're actually giving it any data. We don't want to spuriously return
        // `Ok(0)` when possible as it will cause calls to write_all() to fail.
        // As a result we execute this in a loop to ensure that we try our
        // darndest to write the data.
        loop {
            self.dump()?;

            let before_in = self.data.total_in();
            let ret = self.data.run_vec(buf, &mut self.buf, D::Flush::none());
            let written = (self.data.total_in() - before_in) as usize;
            let is_stream_end = matches!(ret, Ok(Status::StreamEnd));

            if !buf.is_empty() && written == 0 && ret.is_ok() && !is_stream_end {
                continue;
            }
            return match ret {
                Ok(st) => match st {
                    Status::Ok | Status::BufError | Status::StreamEnd => Ok((written, st)),
                },
                Err(..) => Err(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "corrupt deflate stream",
                )),
            };
        }
    }

    fn dump(&mut self) -> io::Result<()> {
        // TODO: should manage this buffer not with `drain` but probably more of
        // a deque-like strategy.
        while !self.buf.is_empty() {
            let n = self.obj.as_mut().unwrap().write(&self.buf)?;
            if n == 0 {
                return Err(io::ErrorKind::WriteZero.into());
            }
            self.buf.drain(..n);
        }
        Ok(())
    }
}

impl<W: Write, D: Ops> Write for Writer<W, D> {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.write_with_status(buf).map(|res| res.0)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.data
            .run_vec(&[], &mut self.buf, D::Flush::sync())
            .unwrap();

        // Unfortunately miniz doesn't actually tell us when we're done with
        // pulling out all the data from the internal stream. To remedy this we
        // have to continually ask the stream for more memory until it doesn't
        // give us a chunk of memory the same size as our own internal buffer,
        // at which point we assume it's reached the end.
        loop {
            self.dump()?;
            let before = self.data.total_out();
            self.data
                .run_vec(&[], &mut self.buf, D::Flush::none())
                .unwrap();
            if before == self.data.total_out() {
                break;
            }
        }

        self.obj.as_mut().unwrap().flush()
    }
}

impl<W: Write, D: Ops> Drop for Writer<W, D> {
    fn drop(&mut self) {
        if self.obj.is_some() {
            let _ = self.finish();
        }
    }
}

Coverage Report

Created: 2024-06-18 07:39

Line	Count	Source (jump to first uncovered line)
1		use std::io;
2		use std::io::prelude::*;
3		use std::mem;
4
5		use crate::{Compress, Decompress, DecompressError, FlushCompress, FlushDecompress, Status};
6
7	0	#[derive(Debug)]
8		pub struct Writer<W: Write, D: Ops> {
9		obj: Option<W>,
10		pub data: D,
11		buf: Vec<u8>,
12		}
13
14		pub trait Ops {
15		type Flush: Flush;
16		fn total_in(&self) -> u64;
17		fn total_out(&self) -> u64;
18		fn run(
19		&mut self,
20		input: &[u8],
21		output: &mut [u8],
22		flush: Self::Flush,
23		) -> Result<Status, DecompressError>;
24		fn run_vec(
25		&mut self,
26		input: &[u8],
27		output: &mut Vec<u8>,
28		flush: Self::Flush,
29		) -> Result<Status, DecompressError>;
30		}
31
32		impl Ops for Compress {
33		type Flush = FlushCompress;
34	0	fn total_in(&self) -> u64 {
35	0	self.total_in()
36	0	}
37	0	fn total_out(&self) -> u64 {
38	0	self.total_out()
39	0	}
40	0	fn run(
41	0	&mut self,
42	0	input: &[u8],
43	0	output: &mut [u8],
44	0	flush: FlushCompress,
45	0	) -> Result<Status, DecompressError> {
46	0	Ok(self.compress(input, output, flush).unwrap())
47	0	}
48	0	fn run_vec(
49	0	&mut self,
50	0	input: &[u8],
51	0	output: &mut Vec<u8>,
52	0	flush: FlushCompress,
53	0	) -> Result<Status, DecompressError> {
54	0	Ok(self.compress_vec(input, output, flush).unwrap())
55	0	}
56		}
57
58		impl Ops for Decompress {
59		type Flush = FlushDecompress;
60	0	fn total_in(&self) -> u64 {
61	0	self.total_in()
62	0	}
63	0	fn total_out(&self) -> u64 {
64	0	self.total_out()
65	0	}
66	0	fn run(
67	0	&mut self,
68	0	input: &[u8],
69	0	output: &mut [u8],
70	0	flush: FlushDecompress,
71	0	) -> Result<Status, DecompressError> {
72	0	self.decompress(input, output, flush)
73	0	}
74	0	fn run_vec(
75	0	&mut self,
76	0	input: &[u8],
77	0	output: &mut Vec<u8>,
78	0	flush: FlushDecompress,
79	0	) -> Result<Status, DecompressError> {
80	0	self.decompress_vec(input, output, flush)
81	0	}
82		}
83
84		pub trait Flush {
85		fn none() -> Self;
86		fn sync() -> Self;
87		fn finish() -> Self;
88		}
89
90		impl Flush for FlushCompress {
91	0	fn none() -> Self {
92	0	FlushCompress::None
93	0	}
94
95	0	fn sync() -> Self {
96	0	FlushCompress::Sync
97	0	}
98
99	0	fn finish() -> Self {
100	0	FlushCompress::Finish
101	0	}
102		}
103
104		impl Flush for FlushDecompress {
105	0	fn none() -> Self {
106	0	FlushDecompress::None
107	0	}
108
109	0	fn sync() -> Self {
110	0	FlushDecompress::Sync
111	0	}
112
113	0	fn finish() -> Self {
114	0	FlushDecompress::Finish
115	0	}
116		}
117
118	0	pub fn read<R, D>(obj: &mut R, data: &mut D, dst: &mut [u8]) -> io::Result<usize>
119	0	where
120	0	R: BufRead,
121	0	D: Ops,
122	0	{
123		loop {
124		let (read, consumed, ret, eof);
125		{
126	0	let input = obj.fill_buf()?;
127	0	eof = input.is_empty();
128	0	let before_out = data.total_out();
129	0	let before_in = data.total_in();
130	0	let flush = if eof {
131	0	D::Flush::finish()
132		} else {
133	0	D::Flush::none()
134		};
135	0	ret = data.run(input, dst, flush);
136	0	read = (data.total_out() - before_out) as usize;
137	0	consumed = (data.total_in() - before_in) as usize;
138	0	}
139	0	obj.consume(consumed);
140
141	0	match ret {
142		// If we haven't ready any data and we haven't hit EOF yet,
143		// then we need to keep asking for more data because if we
144		// return that 0 bytes of data have been read then it will
145		// be interpreted as EOF.
146	0	Ok(Status::Ok \| Status::BufError) if read == 0 && !eof && !dst.is_empty() => continue,
147	0	Ok(Status::Ok \| Status::BufError \| Status::StreamEnd) => return Ok(read),
148
149		Err(..) => {
150	0	return Err(io::Error::new(
151	0	io::ErrorKind::InvalidInput,
152	0	"corrupt deflate stream",
153	0	))
154		}
155		}
156		}
157	0	}
158
159		impl<W: Write, D: Ops> Writer<W, D> {
160	0	pub fn new(w: W, d: D) -> Writer<W, D> {
161	0	Writer {
162	0	obj: Some(w),
163	0	data: d,
164	0	buf: Vec::with_capacity(32 * 1024),
165	0	}
166	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress>>::new Unexecuted instantiation: <flate2::zio::Writer<_, _>>::new
167
168	0	pub fn finish(&mut self) -> io::Result<()> {
169		loop {
170	0	self.dump()?;
171
172	0	let before = self.data.total_out();
173	0	self.data.run_vec(&[], &mut self.buf, D::Flush::finish())?;
174	0	if before == self.data.total_out() {
175	0	return Ok(());
176	0	}
177		}
178	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress>>::finish Unexecuted instantiation: <flate2::zio::Writer<_, _>>::finish
179
180	0	pub fn replace(&mut self, w: W) -> W {
181	0	self.buf.truncate(0);
182	0	mem::replace(self.get_mut(), w)
183	0	}
184
185	0	pub fn get_ref(&self) -> &W {
186	0	self.obj.as_ref().unwrap()
187	0	}
188
189	0	pub fn get_mut(&mut self) -> &mut W {
190	0	self.obj.as_mut().unwrap()
191	0	}
192
193		// Note that this should only be called if the outer object is just about
194		// to be consumed!
195		//
196		// (e.g. an implementation of `into_inner`)
197	0	pub fn take_inner(&mut self) -> W {
198	0	self.obj.take().unwrap()
199	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress>>::take_inner Unexecuted instantiation: <flate2::zio::Writer<_, _>>::take_inner
200
201	0	pub fn is_present(&self) -> bool {
202	0	self.obj.is_some()
203	0	}
204
205		// Returns total written bytes and status of underlying codec
206	0	pub(crate) fn write_with_status(&mut self, buf: &[u8]) -> io::Result<(usize, Status)> {
207		// miniz isn't guaranteed to actually write any of the buffer provided,
208		// it may be in a flushing mode where it's just giving us data before
209		// we're actually giving it any data. We don't want to spuriously return
210		// `Ok(0)` when possible as it will cause calls to write_all() to fail.
211		// As a result we execute this in a loop to ensure that we try our
212		// darndest to write the data.
213		loop {
214	0	self.dump()?;
215
216	0	let before_in = self.data.total_in();
217	0	let ret = self.data.run_vec(buf, &mut self.buf, D::Flush::none());
218	0	let written = (self.data.total_in() - before_in) as usize;
219	0	let is_stream_end = matches!(ret, Ok(Status::StreamEnd));
220
221	0	if !buf.is_empty() && written == 0 && ret.is_ok() && !is_stream_end {
222	0	continue;
223	0	}
224	0	return match ret {
225	0	Ok(st) => match st {
226	0	Status::Ok \| Status::BufError \| Status::StreamEnd => Ok((written, st)),
227		},
228	0	Err(..) => Err(io::Error::new(
229	0	io::ErrorKind::InvalidInput,
230	0	"corrupt deflate stream",
231	0	)),
232		};
233		}
234	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress>>::write_with_status Unexecuted instantiation: <flate2::zio::Writer<_, _>>::write_with_status
235
236	0	fn dump(&mut self) -> io::Result<()> {
237		// TODO: should manage this buffer not with `drain` but probably more of
238		// a deque-like strategy.
239	0	while !self.buf.is_empty() {
240	0	let n = self.obj.as_mut().unwrap().write(&self.buf)?;
241	0	if n == 0 {
242	0	return Err(io::ErrorKind::WriteZero.into());
243	0	}
244	0	self.buf.drain(..n);
245		}
246	0	Ok(())
247	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress>>::dump Unexecuted instantiation: <flate2::zio::Writer<_, _>>::dump
248		}
249
250		impl<W: Write, D: Ops> Write for Writer<W, D> {
251	0	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
252	0	self.write_with_status(buf).map(\|res\| res.0) Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress> as std::io::Write>::write::{closure#0} Unexecuted instantiation: <flate2::zio::Writer<_, _> as std::io::Write>::write::{closure#0}
253	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress> as std::io::Write>::write Unexecuted instantiation: <flate2::zio::Writer<_, _> as std::io::Write>::write
254
255	0	fn flush(&mut self) -> io::Result<()> {
256	0	self.data
257	0	.run_vec(&[], &mut self.buf, D::Flush::sync())
258	0	.unwrap();
259
260		// Unfortunately miniz doesn't actually tell us when we're done with
261		// pulling out all the data from the internal stream. To remedy this we
262		// have to continually ask the stream for more memory until it doesn't
263		// give us a chunk of memory the same size as our own internal buffer,
264		// at which point we assume it's reached the end.
265		loop {
266	0	self.dump()?;
267	0	let before = self.data.total_out();
268	0	self.data
269	0	.run_vec(&[], &mut self.buf, D::Flush::none())
270	0	.unwrap();
271	0	if before == self.data.total_out() {
272	0	break;
273	0	}
274		}
275
276	0	self.obj.as_mut().unwrap().flush()
277	0	}
278		}
279
280		impl<W: Write, D: Ops> Drop for Writer<W, D> {
281	0	fn drop(&mut self) {
282	0	if self.obj.is_some() {
283	0	let _ = self.finish();
284	0	}
285	0	} Unexecuted instantiation: <flate2::zio::Writer<alloc::vec::Vec<u8>, flate2::mem::Compress> as core::ops::drop::Drop>::drop Unexecuted instantiation: <flate2::zio::Writer<_, _> as core::ops::drop::Drop>::drop
286		}