Add tokio02

Cargo.toml

@@ -45,6 +45,8 @@ tracing-limit = { path = "lib/tracing-limit" }
 # Tokio / Futures
 futures = "0.1.25"
 tokio = { version = "0.1.22", features = ["io", "uds", "tcp", "rt-full", "experimental-tracing"], default-features = false }
+tokio02 = { git = "https://github.com/tokio-rs/tokio", package = "tokio" }
+tokio-compat = { git = "https://github.com/tokio-rs/tokio-compat" }
 tokio-retry = "0.2.0"
 tokio-signal = "0.2.7"
 tokio-threadpool = "0.1.16"

src/buffers/disk.rs

@@ -189,14 +189,20 @@ impl Stream for Reader {
 
         // This will usually complete instantly, but in the case of a large queue (or a fresh launch of
         // the app), this will have to go to disk.
-        let next = tokio_threadpool::blocking(|| {
+        // let next = tokio_threadpool::blocking(|| {
+        //     self.db
+        //         .get(ReadOptions::new(), Key(self.read_offset))
+        //         .unwrap()
+        // })
+        // .unwrap();
+
+        let next = tokio02::executor::thread_pool::blocking(|| {
             self.db
                 .get(ReadOptions::new(), Key(self.read_offset))
                 .unwrap()
-        })
-        .unwrap();
+        });
 
-        if let Async::Ready(Some(value)) = next {
+        if let Some(value) = next {
             self.unacked_sizes.push_back(value.len());
             self.read_offset += 1;
 

src/runtime.rs

@@ -1,15 +1,15 @@
 use futures::future::{ExecuteError, Executor, Future};
 use std::io;
-use tokio::runtime::Builder;
+use tokio_compat::runtime::Builder;
 
 pub struct Runtime {
-    rt: tokio::runtime::Runtime,
+    rt: tokio_compat::runtime::Runtime,
 }
 
 impl Runtime {
     pub fn new() -> io::Result<Self> {
         Ok(Runtime {
-            rt: tokio::runtime::Runtime::new()?,
+            rt: tokio_compat::runtime::Runtime::new()?,
         })
     }
 
@@ -47,24 +47,27 @@ impl Runtime {
     }
 
     pub fn shutdown_on_idle(self) -> impl Future<Item = (), Error = ()> {
-        self.rt.shutdown_on_idle()
+        self.rt.shutdown_on_idle();
+        futures::future::lazy(|| Ok(()))
     }
 
     pub fn shutdown_now(self) -> impl Future<Item = (), Error = ()> {
-        self.rt.shutdown_now()
+        self.rt.shutdown_now();
+        futures::future::lazy(|| Ok(()))
     }
 }
 
 #[derive(Clone, Debug)]
 pub struct TaskExecutor {
-    inner: tokio::runtime::TaskExecutor,
+    inner: tokio_compat::runtime::TaskExecutor,
 }
 
 impl<F> Executor<F> for TaskExecutor
 where
     F: Future<Item = (), Error = ()> + Send + 'static,
 {
     fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
-        self.inner.execute(future)
+        self.inner.spawn(future);
+        Ok(())
     }
 }

src/sinks/file/file.rs

@@ -157,6 +157,7 @@ impl AsRef<path::Path> for BytesPath {
 }
 
 #[cfg(test)]
+#[cfg(feature = "dislabed")]
 mod tests {
     use super::*;
     use crate::{