-
Notifications
You must be signed in to change notification settings - Fork 95
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
232 additions
and
8 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1 +1,3 @@ | ||
mod lz4; | ||
|
||
pub mod read; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,131 @@ | ||
use lz4::liblz4::*; | ||
use std::io::{Error, ErrorKind, Result}; | ||
use std::ptr; | ||
use tokio::io::{AsyncRead, ReadBuf}; | ||
|
||
const BUFFER_SIZE: usize = 32 * 1024; | ||
|
||
#[derive(Debug)] | ||
struct DecoderContext { | ||
c: LZ4FDecompressionContext, | ||
} | ||
|
||
#[derive(Debug)] | ||
pub struct Lz4Decoder<R> { | ||
c: DecoderContext, | ||
r: R, | ||
buf: Box<[u8]>, // 32kb input buffer. we repeatedly fill this and decompress into the output buffer. | ||
pos: usize, // cursor into `buf` that represents the start of data in `buf` that has not yet been decompressed. | ||
len: usize, // represents the number of valid bytes in `buf`. | ||
next: usize, // the number of bytes to read into `buf` on the next read. | ||
} | ||
|
||
impl<R: AsyncRead> Lz4Decoder<R> { | ||
/// Creates a new decoder which reads its input from the given | ||
/// input stream. The input stream can be re-acquired by calling | ||
/// `finish()` | ||
pub fn new(r: R) -> Result<Lz4Decoder<R>> { | ||
Ok(Lz4Decoder { | ||
r, | ||
c: DecoderContext::new()?, | ||
buf: vec![0; BUFFER_SIZE].into_boxed_slice(), | ||
pos: BUFFER_SIZE, | ||
len: BUFFER_SIZE, | ||
// Minimal LZ4 stream size | ||
next: 11, | ||
}) | ||
} | ||
|
||
pub fn into_inner(self) -> R { | ||
self.r | ||
} | ||
|
||
pub fn finish(self) -> (R, Result<()>) { | ||
( | ||
self.r, | ||
match self.next { | ||
0 => Ok(()), | ||
_ => Err(Error::new( | ||
ErrorKind::Interrupted, | ||
"Finish runned before read end of compressed stream", | ||
)), | ||
}, | ||
) | ||
} | ||
} | ||
|
||
impl<R: AsyncRead + std::marker::Unpin> AsyncRead for Lz4Decoder<R> { | ||
fn poll_read( | ||
self: std::pin::Pin<&mut Self>, | ||
cx: &mut std::task::Context<'_>, | ||
buf: &mut tokio::io::ReadBuf<'_>, | ||
) -> std::task::Poll<std::io::Result<()>> { | ||
if self.next == 0 || buf.remaining() == 0 { | ||
return std::task::Poll::Ready(Ok(())); | ||
} | ||
let mut written_len: usize = 0; | ||
let mself = self.get_mut(); | ||
while written_len == 0 { | ||
if mself.pos >= mself.len { | ||
Check failure on line 69 in rust_async/src/lz4.rs GitHub Actions / spellcheck
|
||
let need = if mself.buf.len() < mself.next { | ||
mself.buf.len() | ||
} else { | ||
mself.next | ||
}; | ||
{ | ||
let mut comp_buf = ReadBuf::new(&mut mself.buf[..need]); | ||
let result = std::pin::pin!(&mut mself.r).poll_read(cx, &mut comp_buf); | ||
match result { | ||
std::task::Poll::Pending => return result, | ||
std::task::Poll::Ready(Err(_)) => return result, | ||
_ => {} | ||
}; | ||
mself.len = comp_buf.filled().len(); | ||
} | ||
if mself.len == 0 { | ||
break; | ||
} | ||
mself.pos = 0; | ||
mself.next -= mself.len; | ||
} | ||
while (written_len < buf.remaining()) && (mself.pos < mself.len) { | ||
let mut src_size = mself.len - mself.pos; | ||
let mut dst_size = buf.remaining() - written_len; | ||
let len = check_error(unsafe { | ||
LZ4F_decompress( | ||
mself.c.c, | ||
buf.initialize_unfilled().as_mut_ptr(), | ||
&mut dst_size, | ||
mself.buf[mself.pos..].as_ptr(), | ||
&mut src_size, | ||
ptr::null(), | ||
) | ||
})?; | ||
mself.pos += src_size as usize; | ||
written_len += dst_size as usize; | ||
buf.set_filled(written_len); | ||
if len == 0 { | ||
mself.next = 0; | ||
return std::task::Poll::Ready(Ok(())); | ||
} else if mself.next < len { | ||
mself.next = len; | ||
} | ||
} | ||
} | ||
return std::task::Poll::Ready(Ok(())); | ||
} | ||
} | ||
|
||
impl DecoderContext { | ||
fn new() -> Result<DecoderContext> { | ||
let mut context = LZ4FDecompressionContext(ptr::null_mut()); | ||
check_error(unsafe { LZ4F_createDecompressionContext(&mut context, LZ4F_VERSION) })?; | ||
Ok(DecoderContext { c: context }) | ||
} | ||
} | ||
|
||
impl Drop for DecoderContext { | ||
fn drop(&mut self) { | ||
unsafe { LZ4F_freeDecompressionContext(self.c) }; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters