tools/wrapped_reader_stream.rs: only implement Stream (without AsyncRead)

Looks more efficent ...
This commit is contained in:
Dietmar Maurer 2019-01-20 10:41:21 +01:00
parent 8e89d9cafe
commit ff8bdf3b64
1 changed files with 9 additions and 27 deletions

View File

@ -1,10 +1,8 @@
use failure::*; use failure::*;
use tokio_threadpool; use tokio_threadpool;
use tokio::io::{AsyncRead};
use std::io::Read; use std::io::Read;
use futures::Async; use futures::Async;
use futures::stream::Stream; use futures::stream::Stream;
use std::io::ErrorKind::{Other, WouldBlock};
pub struct WrappedReaderStream<R: Read> { pub struct WrappedReaderStream<R: Read> {
reader: R, reader: R,
@ -17,24 +15,10 @@ impl <R: Read> WrappedReaderStream<R> {
} }
} }
impl <R: Read> Read for WrappedReaderStream<R> { fn blocking_err() -> std::io::Error {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> { std::io::Error::new(
//tokio::io::would_block(|| self.reader.read(buf)) std::io::ErrorKind::Other,
// fixme: howto?? "`blocking` annotated I/O must be called from the context of the Tokio runtime.")
match tokio_threadpool::blocking(|| self.reader.read(buf)) {
Ok(Async::Ready(res)) => res,
Ok(Async::NotReady) => Err(WouldBlock.into()),
Err(err) => Err(std::io::Error::new(Other, "`blocking` annotated I/O must be called \
from the context of the Tokio runtime.")),
}
}
}
impl <R: Read> AsyncRead for WrappedReaderStream<R> {
// fixme:???!!?
unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
false
}
} }
impl <R: Read> Stream for WrappedReaderStream<R> { impl <R: Read> Stream for WrappedReaderStream<R> {
@ -44,19 +28,17 @@ impl <R: Read> Stream for WrappedReaderStream<R> {
fn poll(&mut self) -> Result<Async<Option<Vec<u8>>>, std::io::Error> { fn poll(&mut self) -> Result<Async<Option<Vec<u8>>>, std::io::Error> {
let mut buf = [0u8;64*1024]; let mut buf = [0u8;64*1024];
match self.poll_read(&mut buf) { match tokio_threadpool::blocking(|| self.reader.read(&mut buf)) {
Ok(Async::Ready(n)) => { Ok(Async::Ready(Ok(n))) => {
// By convention, if an AsyncRead says that it read 0 bytes, if n == 0 { // EOF
// we should assume that it has got to the end, so we signal that
// the Stream is done in this case by returning None:
if n == 0 {
Ok(Async::Ready(None)) Ok(Async::Ready(None))
} else { } else {
Ok(Async::Ready(Some(buf[..n].to_vec()))) Ok(Async::Ready(Some(buf[..n].to_vec())))
} }
}, },
Ok(Async::Ready(Err(err))) => Err(err),
Ok(Async::NotReady) => Ok(Async::NotReady), Ok(Async::NotReady) => Ok(Async::NotReady),
Err(e) => Err(e) Err(err) => Err(blocking_err()),
} }
} }
} }