Add timeout functionality

- Added timeout future funcitonality
- changes variable names

Author: StaticallyTypedAnxiety

Cargo.toml

@@ -11,6 +11,7 @@ dashmap = "6.1.0"
 lazy_static = "1.5.0"
 wit-bindgen-rt = { version = "0.30.0", features = ["bitflags"] }
 autoincrement = { version = "1", features = ["derive", "async"] }
+pin-project-lite = "0.2.16"
 [dependencies.uuid]
 version = "1.17.0"
 # Lets you generate random UUIDs

src/io/timer.rs

@@ -1,8 +1,10 @@
 use crate::engine::REACTOR;
 use crate::poll_tasks::EventWithWaker;
 use dashmap::DashMap;
+use futures::FutureExt;
 use lazy_static::lazy_static;
 use std::future::Future;
+use std::task::Poll;
 use std::time::Duration;
 use std::time::Instant;
 use uuid::Uuid;
@@ -13,28 +15,45 @@ lazy_static! {
 #[derive(Debug, Clone)]
 pub struct Timer {
     at: Instant,
-    until: Duration,
+    deadline: Duration,
     elapsed: bool,
 }
 
 impl Timer {
     /// create a timer that resolves once it elapses
     pub async fn sleep(until: std::time::Duration) {
-        let timeout = TimeoutFuture {
+        let timeout = TimeFuture {
             timer_key: format!("sleep-{}", Uuid::new_v4()),
             timer: Self {
                 at: Instant::now(),
-                until,
+                deadline: until,
                 elapsed: false,
             },
         };
         timeout.await
     }
+
+    pub async fn timeout<K, F: Future<Output = K>>(
+        &self,
+        fut: F,
+        deadline: std::time::Duration,
+    ) -> std::io::Result<K> {
+        let timer_future = TimeFuture {
+            timer_key: format!("timeout-{}", Uuid::new_v4()),
+            timer: Self {
+                at: Instant::now(),
+                deadline,
+                elapsed: false,
+            },
+        };
+        let timeout_future = TimeoutFuture { timer_future, fut };
+        timeout_future.await
+    }
     pub fn update_elapsed(&mut self) {
         let new_now = Instant::now();
         let elapsed = new_now
             .checked_duration_since(self.at)
-            .map(|s| s > self.until)
+            .map(|s| s > self.deadline)
             .unwrap_or_default();
         self.elapsed = elapsed;
     }
@@ -44,12 +63,12 @@ impl Timer {
     }
 }
 
-struct TimeoutFuture {
+struct TimeFuture {
     timer_key: String,
     timer: Timer,
 }
 
-impl Future for TimeoutFuture {
+impl Future for TimeFuture {
     type Output = ();
     fn poll(
         self: std::pin::Pin<&mut Self>,
@@ -72,3 +91,32 @@ impl Future for TimeoutFuture {
         std::task::Poll::Pending
     }
 }
+
+pin_project_lite::pin_project! {
+    pub struct TimeoutFuture<K,F:Future<Output = K>>
+    {
+        #[pin]
+        fut:F,
+        #[pin]
+        timer_future:TimeFuture
+    }
+}
+
+impl<K, F: Future<Output = K>> Future for TimeoutFuture<K, F> {
+    type Output = Result<K, std::io::Error>;
+    fn poll(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        let mut this = self.project();
+        if this.timer_future.poll_unpin(cx).is_pending() {
+            match this.fut.poll_unpin(cx) {
+                Poll::Ready(ready) => Poll::Ready(Ok(ready)),
+                Poll::Pending => Poll::Pending,
+            }
+        } else {
+            let error = std::io::Error::new(std::io::ErrorKind::TimedOut, "Timer has elapsed");
+            Poll::Ready(Err(error))
+        }
+    }
+}