add stream heartbeat

1 files changed, 21 insertions, 1 deletions

diff --git a/src/api/store.rs b/src/api/store.rs
index eac3262..d419d37 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -861,6 +861,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo
 
     let stream_state = state.clone();
     let stream_key = key.clone();
+    let heartbeat = state.stream_heartbeat;
 
     let output = async_stream::stream! {
         let (mut last_value, mut conflicted) = initial;
@@ -872,9 +873,27 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo
                 return;
             }
         }
+        // When the last line went out: a blank one follows after `heartbeat` without
+        // another, so proxies in front don't take the stream for dead.
+        let mut sent = tokio::time::Instant::now();
 
         loop {
-            let relevant = match changes.recv().await {
+            let beat = async {
+                match sent.checked_add(heartbeat).filter(|_| !heartbeat.is_zero()) {
+                    Some(at) => tokio::time::sleep_until(at).await,
+                    None => std::future::pending::<()>().await,
+                }
+            };
+            let received = tokio::select! {
+                received = changes.recv() => Some(received),
+                () = beat => None,
+            };
+            let Some(received) = received else {
+                sent = tokio::time::Instant::now();
+                yield Ok(Bytes::from_static(b"\n"));
+                continue;
+            };
+            let relevant = match received {
                 // Conflicted paths can change without a patch on their path; re-read always.
                 Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)),
                 Err(broadcast::error::RecvError::Lagged(skipped)) => {
@@ -910,6 +929,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo
                     return;
                 }
             }
+            sent = tokio::time::Instant::now();
             last_value = next_value;
         }
     };