add expiration after insert subscription

fengys1996 · Feb 19, 2025 · 1df1c2f · 1df1c2f
1 parent ccfb138
commit 1df1c2f
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/src/producer/fetch.rs b/src/producer/fetch.rs
@@ -115,6 +115,12 @@ impl ResultFetcher {
 
         let (tx, rx) = channel();
         self.inner.subscriptions.insert(request_id, tx);
+        let expired_at = Instant::now() + Duration::from_secs(5);
+        self.inner
+            .expirations
+            .write()
+            .unwrap()
+            .push_back((request_id, expired_at));
         Ok(rx)
     }