async_read_to_stream.rs

use futures::{future::ok, try_ready, Async, Future, Poll, Stream};
use tokio::io::AsyncRead;

use std::io;

pub struct ByteStream<R: AsyncRead>(R);

impl<R: AsyncRead> ByteStream<R> {
  pub fn new(reader: R) -> Self {
    ByteStream(reader)
  }
}

impl<R: AsyncRead> Stream for ByteStream<R> {
  type Item = Vec<u8>;
  type Error = io::Error;

  fn poll(&mut self) -> Poll<Option<Self::Item>, io::Error> {
    let mut buffer = vec![0; 1518];

    let n = try_ready!(self.0.poll_read(buffer.as_mut()));
    Ok(Async::Ready(Some(buffer[..n].to_vec())))
  }
}

#[test]
fn read_works() {
  // Input followed by Enter
  let reader = tokio::io::stdin();
  let stream = ByteStream(reader);
  let s = stream
    .for_each(|b| {
      print!("Output: {:?}\na", b);
      ok(())
    })
    .map_err(|_| ());
  tokio::run(s);
}