Added ToClientMessage::EventLiveBoundary

Author: spirali

crates/hyperqueue/src/client/commands/journal/mod.rs

@@ -71,11 +71,15 @@ async fn stream_json(gsettings: &GlobalSettings, live_events: bool) -> anyhow::R
     let mut stdout = BufWriter::new(stdout);
     while let Some(event) = connection.connection().receive().await {
         let event = event?;
-        if let ToClientMessage::Event(e) = event {
-            writeln!(stdout, "{}", format_event(e))?;
-            stdout.flush()?;
-        } else {
-            anyhow::bail!("Invalid message receive, not ToClientMessage::Event");
+        match event {
+            ToClientMessage::Event(e) => {
+                writeln!(stdout, "{}", format_event(e))?;
+                stdout.flush()?;
+            }
+            ToClientMessage::EventLiveBoundary => { /* Do nothing */ }
+            _ => {
+                anyhow::bail!("Invalid message receive, not ToClientMessage::Event");
+            }
         }
     }
     Ok(())

crates/hyperqueue/src/dashboard/data/fetch.rs

@@ -35,16 +35,21 @@ pub async fn create_data_fetch_process(
                 let Some(message) = message else { break; };
 
                 let message = message?;
-                let ToClientMessage::Event(event) = message else {
-                    return Err(anyhow::anyhow!(
-                        "Dashboard received unexpected message {message:?}"
-                    ));
+                match message {
+                    ToClientMessage::Event(event) => {
+                        events.push(event);
+                        if events.len() == CAPACITY {
+                            sender.send(events).await?;
+                            events = Vec::with_capacity(CAPACITY);
+                        }
+                    },
+                    ToClientMessage::EventLiveBoundary => {
+                        /* Do nothing */
+                    }
+                    _ => {
+                        return Err(anyhow::anyhow!("Dashboard received unexpected message {message:?}"));
+                    }
                 };
-                events.push(event);
-                if events.len() == CAPACITY {
-                    sender.send(events).await?;
-                    events = Vec::with_capacity(CAPACITY);
-                }
             }
         }
     }

crates/hyperqueue/src/server/client/mod.rs

@@ -185,6 +185,7 @@ pub async fn client_rpc_loop<
                         senders.events.replay_journal(tx1);
                         stream_history_events(&mut tx, rx1).await;
                         if let Some((rx2, listener_id)) = live {
+                            let _ = tx.send(ToClientMessage::EventLiveBoundary).await;
                             stream_events(&mut tx, &mut rx, rx2).await;
                             senders.events.unregister_listener(listener_id);
                         }

crates/hyperqueue/src/transfer/messages.rs

@@ -41,6 +41,7 @@ pub enum FromClientMessage {
     // This command switches the connection into streaming connection,
     // it will no longer reacts to any other client messages
     // and client will only receive ToClientMessage::Event
+    // or ToClientMessage::EventLiveBoundary
     StreamEvents(StreamEvents),
     PruneJournal,
     FlushJournal,
@@ -348,6 +349,9 @@ pub enum ToClientMessage {
     Error(String),
     ServerInfo(ServerInfo),
     Event(Event),
+    // This indicates in live event streaming when old events where
+    // old streamed, and now we are getting new ones
+    EventLiveBoundary,
     Finished, // Generic response, now used only for journal pruning/flushing
 }