feat: add consume_with_watchdog stream helper with tests

src-tauri/src/lib.rs

@@ -5,12 +5,14 @@ mod state;
 mod telemetry;
 mod commands;
 
+use std::pin::Pin;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
 
-use futures::StreamExt;
+use futures::{stream, Stream, StreamExt};
 use serde::Serialize;
 use tauri::{AppHandle, Emitter, Manager, async_runtime};
+use tokio::time;
 
 use crate::ble::{BleSource, MockSource, SensorSource};
 use crate::commands::Command;
@@ -207,6 +209,31 @@ fn trainer_device(state: &State) -> Result<Device, String> {
         .ok_or_else(|| "станок не найден".to_string())
 }
 
+/// Why the watchdog loop stopped consuming a stream.
+enum StreamOutcome {
+    /// The stream ended (peripheral disconnected / notifications closed).
+    Ended,
+    /// No item arrived within `timeout` (stale connection).
+    Stale,
+}
+
+async fn consume_with_watchdog<F>(
+    stream: &mut Pin<Box<dyn Stream<Item = Telemetry> + Send>>,
+    timeout: Duration,
+    mut on_item: F,
+) -> StreamOutcome
+where
+    F: FnMut(Telemetry),
+{
+    loop {
+        match time::timeout(timeout, stream.next()).await {
+            Ok(Some(t)) => on_item(t),
+            Ok(None) => return StreamOutcome::Ended,
+            Err(_) => return StreamOutcome::Stale,
+        }
+    }
+}
+
 #[cfg_attr(mobile, tauri::mobile_entry_point)]
 pub fn run() {
     tauri::Builder::default()
@@ -243,3 +270,45 @@ pub fn run() {
         .run(tauri::generate_context!())
         .expect("error while running tauri application");
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::stream;
+    use std::time::Duration;
+
+    fn pin(s: impl Stream<Item = Telemetry> + Send + 'static) -> Pin<Box<dyn Stream<Item = Telemetry> + Send>> {
+        Box::pin(s)
+    }
+
+    #[tokio::test]
+    async fn ends_when_stream_is_done() {
+        let items = vec![Telemetry::HeartRate { bpm: 120 }];
+        let mut s = pin(stream::iter(items));
+        let mut seen = Vec::new();
+        let outcome = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await;
+        assert!(matches!(outcome, StreamOutcome::Ended));
+        assert_eq!(seen, vec![Telemetry::HeartRate { bpm: 120 }]);
+    }
+
+    #[tokio::test]
+    async fn stale_when_silent_past_timeout() {
+        let mut s = pin(stream::pending::<Telemetry>());
+        let mut seen = Vec::new();
+        let outcome = consume_with_watchdog(&mut s, Duration::from_millis(50), |t| seen.push(t)).await;
+        assert!(matches!(outcome, StreamOutcome::Stale));
+        assert!(seen.is_empty());
+    }
+
+    #[tokio::test]
+    async fn forwards_every_item_via_callback() {
+        let items = vec![
+            Telemetry::Power { watts: 200 },
+            Telemetry::Cadence { rpm: 90.0 },
+        ];
+        let mut s = pin(stream::iter(items));
+        let mut seen = Vec::new();
+        let _ = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await;
+        assert_eq!(seen.len(), 2);
+    }
+}