Port tor_rtcompat::timer to use Runtime objects.

tor-rtcompat/src/impls/async_std.rs

@@ -27,32 +27,10 @@ pub mod net {
 }
 
 /// Functions for launching and managing tasks (async_std implementation)
-pub mod task {
+pub mod task {}
-    pub use async_std_crate::task::{block_on, sleep, spawn, JoinHandle};
-
-    //pub use async_std_crate::task::JoinHandle;
-
-    /// Stop the task `handle` from running.
-    ///
-    /// If you drop `handle` without calling this function, it will just
-    /// run to completion.
-    #[allow(unused)]
-    pub async fn cancel_task<T>(handle: JoinHandle<T>) {
-        handle.cancel().await;
-    }
-}
 
 /// Functions and types for manipulating timers (async_std implementation)
-pub mod timer {
+pub mod timer {}
-    use std::time::Duration;
-
-    /// Return a future that will be ready after `duration` has passed.
-    pub fn sleep(duration: Duration) -> async_io::Timer {
-        async_io::Timer::after(duration)
-    }
-
-    pub use async_std_crate::future::{timeout, TimeoutError};
-}
 
 /// Implement TLS using async_std and async_native_tls.
 pub mod tls {

tor-rtcompat/src/impls/tokio.rs

@@ -130,9 +130,7 @@ pub mod net {
 pub mod task {}
 
 /// Functions and types for manipulating timers (tokio implementation)
-pub mod timer {
+pub mod timer {}
-    pub use tokio_crate::time::{error::Elapsed as TimeoutError, sleep, timeout};
-}
 
 /// Implement a set of TLS wrappers for use with tokio.
 ///

tor-rtcompat/src/lib.rs

@@ -89,9 +89,64 @@ pub mod task {
 
 /// Functions and types for manipulating timers.
 pub mod timer {
-    use std::time::{Duration, SystemTime};
+    use futures::Future;
+    use pin_project::pin_project;
+    use std::{
+        pin::Pin,
+        task::{Context, Poll},
+        time::{Duration, SystemTime},
+    };
 
-    pub use crate::imp::timer::*;
+    pub use crate::task::sleep; // XXXX redundant.
+
+    #[derive(Copy, Clone, Debug)]
+    pub struct TimeoutError;
+    impl std::error::Error for TimeoutError {}
+    impl std::fmt::Display for TimeoutError {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "Timeout expired")
+        }
+    }
+
+    #[pin_project]
+    pub struct Timeout<T, S> {
+        #[pin]
+        future: T,
+        #[pin]
+        sleep_future: S,
+    }
+
+    pub fn timeout<F: Future>(
+        duration: Duration,
+        future: F,
+    ) -> impl Future<Output = Result<F::Output, TimeoutError>> {
+        let sleep_future = crate::task::sleep(duration);
+
+        Timeout {
+            future,
+            sleep_future,
+        }
+    }
+
+    impl<T, S> Future for Timeout<T, S>
+    where
+        T: Future,
+        S: Future<Output = ()>,
+    {
+        type Output = Result<T::Output, TimeoutError>;
+
+        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            let this = self.project();
+            if let Poll::Ready(x) = this.future.poll(cx) {
+                return Poll::Ready(Ok(x));
+            }
+
+            match this.sleep_future.poll(cx) {
+                Poll::Pending => Poll::Pending,
+                Poll::Ready(()) => Poll::Ready(Err(TimeoutError)),
+            }
+        }
+    }
 
     /// Pause until the wall-clock is at `when` or later, trying to
     /// recover from clock jumps.