use std::io::{
Error,
ErrorKind,
Result,
SeekFrom,
};
use crate::buffered::DEFAULT_BUFFER_CAPACITY;
use crate::traits::validate_read_count;
use crate::util::UncheckedSlice;
use crate::{
Buffer,
Input,
Seekable,
SeekableInput,
};
#[derive(Debug)]
pub struct BufferedInput<I>
where
I: Input,
I::Item: Copy + Default,
{
inner: I,
buffer: Buffer<I::Item>,
}
impl<I> BufferedInput<I>
where
I: Input,
I::Item: Copy + Default,
{
#[inline(always)]
#[must_use]
pub fn new(inner: I) -> Self {
Self::with_capacity(inner, DEFAULT_BUFFER_CAPACITY)
}
#[inline]
#[must_use]
pub fn with_capacity(inner: I, capacity: usize) -> Self {
Self {
inner,
buffer: Buffer::with_capacity(capacity),
}
}
#[inline(always)]
pub const fn inner(&self) -> &I {
&self.inner
}
#[inline(always)]
pub fn inner_mut(&mut self) -> &mut I {
&mut self.inner
}
#[inline(always)]
#[must_use]
pub fn into_inner(self) -> I {
self.inner
}
#[inline(always)]
#[must_use]
pub fn into_parts(self) -> (I, Buffer<I::Item>) {
(self.inner, self.buffer)
}
#[inline(always)]
#[must_use]
pub fn capacity(&self) -> usize {
self.buffer.capacity()
}
#[inline(always)]
#[must_use]
pub fn unread_len(&self) -> usize {
self.buffer.available()
}
#[inline(always)]
#[must_use]
pub fn unread(&self) -> &[I::Item] {
self.buffer.readable()
}
#[inline(always)]
pub unsafe fn consume(&mut self, count: usize) {
unsafe {
self.buffer.consume(count);
}
}
#[inline]
pub unsafe fn copy_unread_to(
&self,
output: &mut [I::Item],
output_index: usize,
count: usize,
) {
debug_assert!(
UncheckedSlice::range_fits(output.len(), output_index, count),
"unchecked unread copy output range exceeds destination buffer",
);
debug_assert!(
count <= self.unread_len(),
"unchecked unread copy exceeds available input buffer",
);
unsafe {
UncheckedSlice::copy_nonoverlapping(
self.buffer.readable(),
0,
output,
output_index,
count,
);
}
}
pub fn fill_more(&mut self) -> Result<bool> {
if self.unread_len() == 0 {
self.discard_buffer();
} else if self.tail_capacity() == 0 {
self.backshift();
if self.tail_capacity() == 0 {
return Err(Error::new(
ErrorKind::InvalidInput,
"buffered input is full; consume buffered items before refilling",
));
}
}
self.read_more()
}
pub fn fill_until(&mut self, count: usize) -> Result<bool> {
if count > self.capacity() {
return Err(Error::new(
ErrorKind::InvalidInput,
"requested available items exceed buffered input capacity",
));
}
while self.unread_len() < count {
let available = self.unread_len();
if available == 0 {
self.discard_buffer();
} else {
let missing = count - available;
if self.tail_capacity() < missing {
self.backshift();
}
}
if !self.read_more()? {
return Ok(false);
}
}
Ok(true)
}
pub fn ensure_available(&mut self, count: usize) -> Result<()> {
if self.fill_until(count)? {
return Ok(());
}
let available = self.unread_len();
unsafe {
self.consume(available);
}
Err(Error::new(
ErrorKind::UnexpectedEof,
"failed to fill whole buffer",
))
}
pub unsafe fn read_unchecked(
&mut self,
output: &mut [I::Item],
output_index: usize,
count: usize,
) -> Result<usize> {
debug_assert!(
UncheckedSlice::range_fits(output.len(), output_index, count),
"unchecked read output range exceeds destination buffer"
);
if count == 0 {
return Ok(0);
}
if self.unread_len() == 0 {
self.discard_buffer();
if count >= self.buffer.capacity() {
let read = unsafe {
self.inner.read_unchecked(output, output_index, count)
}?;
validate_read_count(read, count)?;
return Ok(read);
}
if !self.read_more()? {
return Ok(0);
}
}
let read_count = count.min(self.unread_len());
unsafe {
self.buffer.copy_to(output, output_index, read_count);
}
Ok(read_count)
}
#[inline(always)]
pub fn read(&mut self, output: &mut [I::Item]) -> Result<usize> {
unsafe { self.read_unchecked(output, 0, output.len()) }
}
pub unsafe fn read_fully_unchecked(
&mut self,
output: &mut [I::Item],
output_index: usize,
count: usize,
) -> Result<usize> {
debug_assert!(
UncheckedSlice::range_fits(output.len(), output_index, count),
"unchecked read-fully output range exceeds destination buffer"
);
if count == 0 {
return Ok(0);
}
let available = self.unread_len();
if available >= count {
unsafe {
self.buffer.copy_to(output, output_index, count);
}
return Ok(count);
}
let mut total = 0;
if available > 0 {
unsafe {
self.buffer.copy_to(output, output_index, available);
}
total = available;
}
let remaining = count - total;
if remaining >= self.buffer.capacity() {
self.discard_buffer();
loop {
match unsafe {
self.inner.read_fully_unchecked(
output,
output_index + total,
remaining,
)
} {
Ok(read) => {
validate_read_count(read, remaining)?;
return Ok(total + read);
}
Err(error) if error.kind() == ErrorKind::Interrupted => {
continue;
}
Err(error) => return Err(error),
}
}
}
while total < count {
let remaining = count - total;
match unsafe {
self.read_unchecked(output, output_index + total, remaining)
} {
Ok(0) => break,
Ok(read) => {
total += read;
}
Err(error) => return Err(error),
}
}
Ok(total)
}
#[inline(always)]
pub fn read_fully(&mut self, output: &mut [I::Item]) -> Result<usize> {
unsafe { self.read_fully_unchecked(output, 0, output.len()) }
}
pub fn seek_to(&mut self, position: SeekFrom) -> Result<u64>
where
I: SeekableInput,
{
match position {
SeekFrom::Current(offset) => {
if self.seek_within_buffer(offset) {
return self.stream_position();
}
let position = self.seek_relative_slow(offset)?;
self.discard_buffer();
Ok(position)
}
other => {
let position = Seekable::seek_to(&mut self.inner, other)?;
self.discard_buffer();
Ok(position)
}
}
}
pub fn stream_position(&mut self) -> Result<u64>
where
I: SeekableInput,
{
let position =
Seekable::seek_to(&mut self.inner, SeekFrom::Current(0))?;
let unread = self.unread_len() as u64;
position.checked_sub(unread).ok_or_else(|| {
Error::new(
ErrorKind::InvalidData,
"buffered unread items exceed wrapped input position",
)
})
}
fn seek_relative_slow(&mut self, offset: i64) -> Result<u64>
where
I: SeekableInput,
{
let unread = self.unread_len() as i64;
let adjusted = offset.checked_sub(unread).ok_or_else(|| {
Error::new(
ErrorKind::InvalidInput,
"current seek offset underflows after buffered adjustment",
)
})?;
Seekable::seek_to(&mut self.inner, SeekFrom::Current(adjusted))
}
fn seek_within_buffer(&mut self, offset: i64) -> bool {
if offset >= 0 {
let count = offset as u64;
if count <= self.unread_len() as u64 {
let count = count as usize;
unsafe {
self.buffer.consume(count);
}
return true;
}
return false;
}
let count = offset.unsigned_abs();
if count <= self.buffer.position() as u64 {
let count = count as usize;
unsafe {
self.buffer.rewind(count);
}
return true;
}
false
}
#[inline(always)]
fn tail_capacity(&self) -> usize {
self.buffer.spare_capacity()
}
#[inline(always)]
fn discard_buffer(&mut self) {
self.buffer.clear();
}
#[inline(always)]
fn backshift(&mut self) {
self.buffer.compact();
}
fn read_more(&mut self) -> Result<bool> {
let count = self.tail_capacity();
debug_assert!(count > 0, "buffer has no tail capacity");
loop {
let limit = self.buffer.limit();
match unsafe {
self.inner
.read_unchecked(self.buffer.data_mut(), limit, count)
} {
Ok(0) => return Ok(false),
Ok(read) => {
validate_read_count(read, count)?;
unsafe {
self.buffer.advance(read);
}
return Ok(true);
}
Err(error) if error.kind() == ErrorKind::Interrupted => {
continue;
}
Err(error) => return Err(error),
}
}
}
}
impl<I> Input for BufferedInput<I>
where
I: Input,
I::Item: Copy + Default,
{
type Item = I::Item;
#[inline(always)]
fn is_buffered(&self) -> bool {
true
}
#[inline(always)]
unsafe fn read_unchecked(
&mut self,
output: &mut [I::Item],
output_index: usize,
count: usize,
) -> Result<usize> {
unsafe {
BufferedInput::read_unchecked(self, output, output_index, count)
}
}
#[inline(always)]
fn read(&mut self, output: &mut [I::Item]) -> Result<usize> {
BufferedInput::read(self, output)
}
#[inline(always)]
unsafe fn read_fully_unchecked(
&mut self,
output: &mut [Self::Item],
index: usize,
count: usize,
) -> Result<usize> {
unsafe {
BufferedInput::read_fully_unchecked(self, output, index, count)
}
}
#[inline(always)]
fn read_fully(&mut self, output: &mut [Self::Item]) -> Result<usize> {
BufferedInput::read_fully(self, output)
}
}
impl<I> Seekable for BufferedInput<I>
where
I: SeekableInput,
<I as Input>::Item: Copy + Default,
{
type Item = <I as Input>::Item;
#[inline(always)]
fn seek_to(&mut self, position: SeekFrom) -> Result<u64> {
BufferedInput::seek_to(self, position)
}
}