src/tools/wrapped_reader_stream.rs: switch to async

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>

src/tools/wrapped_reader_stream.rs

@@ -1,15 +1,16 @@
-//use failure::*;
+use std::io::{self, Read};
-use tokio_threadpool;
+use std::pin::Pin;
-use std::io::Read;
+use std::task::{Context, Poll};
-use futures::Async;
+
+use tokio_executor::threadpool::blocking;
 use futures::stream::Stream;
 
-pub struct WrappedReaderStream<R: Read> {
+pub struct WrappedReaderStream<R: Read + Unpin> {
     reader: R,
     buffer: Vec<u8>,
 }
 
-impl <R: Read> WrappedReaderStream<R> {
+impl <R: Read + Unpin> WrappedReaderStream<R> {
 
     pub fn new(reader: R) -> Self {
         let mut buffer = Vec::with_capacity(64*1024);
@@ -18,29 +19,26 @@ impl <R: Read> WrappedReaderStream<R> {
     }
 }
 
-fn blocking_err() -> std::io::Error {
+impl<R: Read + Unpin> Stream for WrappedReaderStream<R> {
-    std::io::Error::new(
+    type Item = Result<Vec<u8>, io::Error>;
-        std::io::ErrorKind::Other,
-        "`blocking` annotated I/O must be called from the context of the Tokio runtime.")
-}
 
-impl <R: Read> Stream for WrappedReaderStream<R> {
+    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Option<Self::Item>> {
-
+        let this = self.get_mut();
-    type Item = Vec<u8>;
+        match blocking(|| this.reader.read(&mut this.buffer)) {
-    type Error = std::io::Error;
+            Poll::Ready(Ok(Ok(n))) => {
-
+                if n == 0 {
-    fn poll(&mut self) -> Result<Async<Option<Vec<u8>>>, std::io::Error> {
+                    // EOF
-        match tokio_threadpool::blocking(|| self.reader.read(&mut self.buffer)) {
+                    Poll::Ready(None)
-            Ok(Async::Ready(Ok(n))) => {
-                 if n == 0 { // EOF
-                    Ok(Async::Ready(None))
                 } else {
-                    Ok(Async::Ready(Some(self.buffer[..n].to_vec())))
+                    Poll::Ready(Some(Ok(this.buffer[..n].to_vec())))
                 }
-            },
+            }
-            Ok(Async::Ready(Err(err))) => Err(err),
+            Poll::Ready(Ok(Err(err))) => Poll::Ready(Some(Err(err))),
-            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(io::Error::new(
-            Err(_) => Err(blocking_err()),
+                io::ErrorKind::Other,
+                err.to_string(),
+            )))),
+            Poll::Pending => Poll::Pending,
         }
     }
 }