use std::collections::TryReserveError;
use std::io::Error;
use std::io::ErrorKind;
use std::io::Result;
use std::io::SeekFrom;
use std::mem::ManuallyDrop;
use std::ptr;
use crate::Buffer;
use crate::Output;
use crate::Seekable;
use crate::SeekableOutput;
use crate::buffered::DEFAULT_BUFFER_CAPACITY;
use crate::buffered::EnsuredBufferedOutput;
use crate::traits::validate_write_count;
#[must_use]
#[derive(Debug)]
pub struct BufferedOutput<O>
where
O: Output,
O::Item: Clone + Default,
{
inner: O,
buffer: Buffer<O::Item>,
panicked: bool,
}
impl<O> BufferedOutput<O>
where
O: Output,
O::Item: Clone + Default,
{
#[inline(always)]
pub fn new(inner: O) -> Self {
Self::with_capacity(inner, DEFAULT_BUFFER_CAPACITY)
}
#[inline]
pub fn with_capacity(inner: O, capacity: usize) -> Self {
Self {
inner,
buffer: Buffer::with_capacity(capacity),
panicked: false,
}
}
#[inline]
pub fn try_with_capacity(inner: O, capacity: usize) -> std::result::Result<Self, TryReserveError> {
Ok(Self {
inner,
buffer: Buffer::try_with_capacity(capacity)?,
panicked: false,
})
}
#[inline]
pub fn ensure(output: O) -> EnsuredBufferedOutput<O> {
if output.is_buffered() {
EnsuredBufferedOutput::AlreadyBuffered(output)
} else {
EnsuredBufferedOutput::Buffered(Self::new(output))
}
}
#[inline]
#[must_use]
pub fn ensure_boxed<'a>(output: O) -> Box<dyn Output<Item = O::Item> + 'a>
where
O: 'a,
O::Item: 'a,
{
if output.is_buffered() {
Box::new(output)
} else {
Box::new(Self::new(output))
}
}
#[inline(always)]
#[must_use]
pub const fn inner(&self) -> &O {
&self.inner
}
#[inline(always)]
#[must_use]
pub fn inner_mut(&mut self) -> &mut O {
&mut self.inner
}
#[inline]
#[must_use = "the returned inner output and pending buffer must be handled"]
pub fn into_parts(self) -> (O, Buffer<O::Item>) {
let this = ManuallyDrop::new(self);
unsafe {
let inner = ptr::read(&this.inner);
let buffer = ptr::read(&this.buffer);
(inner, buffer)
}
}
#[inline(always)]
#[must_use]
pub fn capacity(&self) -> usize {
self.buffer.capacity()
}
#[inline(always)]
pub fn try_reserve_capacity(
&mut self,
capacity: usize,
) -> std::result::Result<(), std::collections::TryReserveError> {
self.buffer.try_reserve_capacity(capacity)
}
#[inline(always)]
#[must_use]
pub fn spare_capacity(&self) -> usize {
self.buffer.spare_capacity()
}
#[inline(always)]
#[must_use]
pub fn spare_raw_parts_mut(&mut self) -> (&mut [O::Item], usize, usize) {
self.buffer.spare_raw_parts_mut()
}
#[inline(always)]
pub unsafe fn advance(&mut self, count: usize) {
unsafe {
self.buffer.advance(count);
}
}
#[inline]
pub fn ensure_spare_capacity(&mut self, count: usize) -> Result<()> {
if count > self.buffer.capacity() {
return Err(Error::new(
ErrorKind::InvalidInput,
"requested spare capacity exceeds buffered output capacity",
));
}
if self.spare_capacity() < count {
self.flush_buffer()?;
}
Ok(())
}
#[inline]
pub unsafe fn write_unchecked(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<usize> {
if count < self.spare_capacity() {
unsafe {
self.write_to_buffer(input, input_index, count);
}
Ok(count)
} else {
unsafe { self.write_cold(input, input_index, count) }
}
}
#[inline]
pub fn write(&mut self, input: &[O::Item]) -> Result<usize> {
let written = unsafe { self.write_unchecked(input, 0, input.len()) }?;
validate_write_count(written, input.len())?;
Ok(written)
}
#[inline]
pub unsafe fn write_fully_unchecked(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<()> {
if count < self.spare_capacity() {
unsafe {
self.write_to_buffer(input, input_index, count);
}
Ok(())
} else {
unsafe { self.write_fully_cold(input, input_index, count) }
}
}
#[inline(always)]
pub fn write_fully(&mut self, input: &[O::Item]) -> Result<()> {
unsafe { self.write_fully_unchecked(input, 0, input.len()) }
}
#[inline]
pub fn flush(&mut self) -> Result<()> {
self.flush_buffer().and_then(|()| Output::flush(&mut self.inner))
}
#[inline]
pub fn stream_position(&mut self) -> Result<u64>
where
O: SeekableOutput,
{
let position = Seekable::seek_to(&mut self.inner, SeekFrom::Current(0))?;
position.checked_add(self.buffer.available() as u64).ok_or_else(|| {
Error::new(
ErrorKind::InvalidData,
"buffered pending items overflow wrapped output position",
)
})
}
#[inline]
pub fn seek_to(&mut self, position: SeekFrom) -> Result<u64>
where
O: SeekableOutput,
{
match position {
SeekFrom::Current(0) => self.stream_position(),
other => self
.flush_buffer()
.and_then(|()| Seekable::seek_to(&mut self.inner, other)),
}
}
fn flush_buffer(&mut self) -> Result<()> {
while !self.buffer.is_empty() {
let position = self.buffer.position();
let available = self.buffer.available();
self.panicked = true;
let result = unsafe { self.inner.write_unchecked(self.buffer.data(), position, available) };
self.panicked = false;
match result {
Ok(0) => {
self.buffer.compact();
return Err(Error::new(ErrorKind::WriteZero, "failed to write buffered data"));
}
Ok(written) => {
if let Err(error) = validate_write_count(written, available) {
self.buffer.compact();
return Err(error);
}
unsafe {
self.buffer.consume(written);
}
}
Err(error) if error.kind() == ErrorKind::Interrupted => {}
Err(error) => {
self.buffer.compact();
return Err(error);
}
}
}
self.buffer.clear();
Ok(())
}
#[inline]
unsafe fn write_to_buffer(&mut self, input: &[O::Item], input_index: usize, count: usize) {
unsafe {
self.buffer.copy_from(input, input_index, count);
}
}
#[inline]
unsafe fn write_inner(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<usize> {
let written = unsafe { self.inner.write_unchecked(input, input_index, count) }?;
validate_write_count(written, count)?;
Ok(written)
}
unsafe fn write_fully_inner(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<()> {
let mut written = 0;
while written < count {
let remaining = count - written;
match unsafe { self.write_inner(input, input_index + written, remaining) } {
Ok(0) => {
return Err(Error::new(ErrorKind::WriteZero, "failed to write whole buffer"));
}
Ok(count) => written += count,
Err(error) if error.kind() == ErrorKind::Interrupted => {}
Err(error) => return Err(error),
}
}
Ok(())
}
#[cold]
#[inline(never)]
unsafe fn write_fully_cold(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<()> {
if count > self.spare_capacity() {
self.flush_buffer()?;
}
if count >= self.buffer.capacity() {
unsafe { self.write_fully_inner(input, input_index, count) }
} else {
unsafe {
self.write_to_buffer(input, input_index, count);
}
Ok(())
}
}
#[cold]
#[inline(never)]
unsafe fn write_cold(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<usize> {
if count > self.spare_capacity() {
self.flush_buffer()?;
}
if count >= self.buffer.capacity() {
unsafe { self.write_inner(input, input_index, count) }
} else {
unsafe {
self.write_to_buffer(input, input_index, count);
}
Ok(count)
}
}
}
impl<O> Output for BufferedOutput<O>
where
O: Output,
O::Item: Clone + Default,
{
type Item = O::Item;
#[inline(always)]
fn is_buffered(&self) -> bool {
true
}
#[inline(always)]
unsafe fn write_unchecked(&mut self, input: &[O::Item], input_index: usize, count: usize) -> Result<usize> {
unsafe { BufferedOutput::write_unchecked(self, input, input_index, count) }
}
#[inline(always)]
fn write(&mut self, input: &[Self::Item]) -> Result<usize> {
BufferedOutput::write(self, input)
}
#[inline(always)]
unsafe fn write_fully_unchecked(&mut self, input: &[Self::Item], index: usize, count: usize) -> Result<()> {
unsafe { BufferedOutput::write_fully_unchecked(self, input, index, count) }
}
#[inline(always)]
fn write_fully(&mut self, input: &[Self::Item]) -> Result<()> {
BufferedOutput::write_fully(self, input)
}
#[inline(always)]
fn flush(&mut self) -> Result<()> {
BufferedOutput::flush(self)
}
}
impl<O> Seekable for BufferedOutput<O>
where
O: SeekableOutput,
<O as Output>::Item: Clone + Default,
{
type Unit = <O as Output>::Item;
#[inline(always)]
fn seek_to(&mut self, position: SeekFrom) -> Result<u64> {
BufferedOutput::seek_to(self, position)
}
}
impl<O> Drop for BufferedOutput<O>
where
O: Output,
O::Item: Clone + Default,
{
fn drop(&mut self) {
if !self.panicked {
drop(self.flush_buffer());
}
}
}