blob: ea71ff643ddf972a953cdf51771913e160014aed [file] [log] [blame]
use futures_core::future::Future;
use futures_core::task::{self, Poll};
use futures_io::{AsyncRead, AsyncWrite};
use std::boxed::Box;
use std::io;
use std::marker::Unpin;
use std::mem::PinMut;
/// A future which will copy all data from a reader into a writer.
///
/// Created by the [`copy_into`] function, this future will resolve to the number of
/// bytes copied or an error if one happens.
///
/// [`copy_into`]: fn.copy_into.html
#[derive(Debug)]
pub struct CopyInto<'a, R: ?Sized + 'a, W: ?Sized + 'a> {
reader: &'a mut R,
read_done: bool,
writer: &'a mut W,
pos: usize,
cap: usize,
amt: u64,
buf: Box<[u8]>,
}
// No projections of PinMut<CopyInto> into PinMut<Field> are ever done.
impl<R: ?Sized, W: ?Sized> Unpin for CopyInto<'_, R, W> {}
impl<'a, R: ?Sized, W: ?Sized> CopyInto<'a, R, W> {
pub(super) fn new(reader: &'a mut R, writer: &'a mut W) -> Self {
CopyInto {
reader,
read_done: false,
writer,
amt: 0,
pos: 0,
cap: 0,
buf: Box::new([0; 2048]),
}
}
}
impl<R, W> Future for CopyInto<'_, R, W>
where R: AsyncRead + ?Sized,
W: AsyncWrite + ?Sized,
{
type Output = io::Result<u64>;
fn poll(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Self::Output> {
let this = &mut *self;
loop {
// If our buffer is empty, then we need to read some data to
// continue.
if this.pos == this.cap && !this.read_done {
let n = try_ready!(this.reader.poll_read(cx, &mut this.buf));
if n == 0 {
this.read_done = true;
} else {
this.pos = 0;
this.cap = n;
}
}
// If our buffer has some data, let's write it out!
while this.pos < this.cap {
let i = try_ready!(this.writer.poll_write(cx, &this.buf[this.pos..this.cap]));
if i == 0 {
return Poll::Ready(Err(
io::Error::new(
io::ErrorKind::WriteZero, "write zero byte into writer")));
} else {
this.pos += i;
this.amt += i as u64;
}
}
// If we've written al the data and we've seen EOF, flush out the
// data and finish the transfer.
// done with the entire transfer.
if this.pos == this.cap && this.read_done {
try_ready!(this.writer.poll_flush(cx));
return Poll::Ready(Ok(this.amt));
}
}
}
}