graph: Refactor Compat::new

Author: leoyvens

graph/src/components/store.rs

@@ -1,7 +1,6 @@
 use failure::Error;
 use futures::stream::poll_fn;
 use futures::{Async, Future, Poll, Stream};
-use futures03::compat::Compat;
 use lazy_static::lazy_static;
 use mockall::predicate::*;
 use mockall::*;
@@ -578,7 +577,7 @@ where
         let mut pending_event: Option<StoreEvent> = None;
         let mut source = self.source.fuse();
         let mut had_err = false;
-        let mut delay = Compat::new(tokio::time::delay_for(interval).map(Result::<_, ()>::Ok));
+        let mut delay = tokio::time::delay_for(interval).unit_error().compat();
         let logger = logger.clone();
 
         let source = Box::new(poll_fn(move || -> Poll<Option<StoreEvent>, ()> {
@@ -605,7 +604,7 @@ where
                 // Timer errors are harmless. Treat them as if the timer had
                 // become ready.
                 Ok(Async::Ready(())) | Err(_) => {
-                    delay = Compat::new(tokio::time::delay_for(interval).map(Ok));
+                    delay = tokio::time::delay_for(interval).unit_error().compat();
                     true
                 }
             };

graph/src/util/futures.rs

@@ -151,7 +151,6 @@ where
         F: Fn() -> R + Send,
         R: Future<Item = I, Error = E> + Send,
     {
-        use futures03::compat::Compat;
         use futures03::future::TryFutureExt;
 
         let operation_name = self.inner.operation_name;
@@ -170,12 +169,11 @@ where
             log_after,
             limit_opt,
             move || {
-                Compat::new(
-                    try_it()
-                        .timeout(timeout)
-                        .map_err(|_| TimeoutError::Elapsed)
-                        .and_then(|res| futures03::future::ready(res.map_err(TimeoutError::Inner))),
-                )
+                try_it()
+                    .timeout(timeout)
+                    .map_err(|_| TimeoutError::Elapsed)
+                    .and_then(|res| futures03::future::ready(res.map_err(TimeoutError::Inner)))
+                    .compat()
             },
         )
     }