timer

Cargo.toml

@@ -94,7 +94,7 @@ fluvio-test-derive = { path = "async-test-derive", version = "0.1.1", optional =
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 async-native-tls = { version = "0.5.0", optional = true }
 async-fs = { version = "2.1", optional = true }
-async-io = { version = "2.3", optional = true }
+async-io = { version = "2.3.3", optional = true }
 async-net = { version = "2.0", optional = true }
 async-std = { version = "1.12", default-features = false, optional = true }
 futures-rustls = { version = "0.26.0", optional = true }

src/timer.rs

@@ -14,18 +14,31 @@ mod inner {
 
     /// same as `after` but return () to make it compatible as previous
     pub fn sleep(duration: Duration) -> Sleeper {
-        Sleeper(after(duration))
+        Sleeper {
+            timer: after(duration),
+            has_fired: false,
+        }
     }
 
     #[pin_project]
-    pub struct Sleeper(#[pin] Timer);
+    pub struct Sleeper {
+        #[pin]
+        timer: Timer,
+        has_fired: bool,
+    }
 
     impl Future for Sleeper {
         type Output = ();
 
         fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
             let this = self.project();
-            if this.0.poll(cx).is_ready() {
+
+            if *this.has_fired {
+                return Poll::Ready(());
+            }
+
+            if this.timer.poll(cx).is_ready() {
+                *this.has_fired = true;
                 Poll::Ready(())
             } else {
                 Poll::Pending