Code review updates

Author: spirali

crates/hyperqueue/src/bin/hq.rs

@@ -46,7 +46,7 @@ use hyperqueue::common::utils::fs::absolute_path;
 use hyperqueue::server::bootstrap::get_client_session;
 use hyperqueue::server::event::streamer::{EventFilter, EventFilterFlags};
 use hyperqueue::transfer::messages::{
-    FromClientMessage, JobInfoRequest, StreamEvents, ToClientMessage,
+    FromClientMessage, JobInfoRequest, StreamEvents, StreamEventsMode, ToClientMessage,
 };
 use hyperqueue::worker::hwdetect::{
     detect_additional_resources, detect_cpus, prune_hyper_threading,
@@ -157,8 +157,7 @@ async fn command_job_wait(gsettings: &GlobalSettings, opts: JobWaitOpts) -> anyh
             selector: opts.selector,
             include_running_tasks: true
         }, Some(StreamEvents {
-            past_events: false,
-            live_events: true,
+            mode: StreamEventsMode::LiveEvents,
             enable_worker_overviews: false,
             filter: EventFilter::new(None, EventFilterFlags::JOB_EVENTS)
         })),
@@ -182,8 +181,7 @@ async fn command_job_progress(
             selector: opts.selector,
             include_running_tasks: true
         }, Some(StreamEvents {
-            past_events: false,
-            live_events: true,
+            mode: StreamEventsMode::LiveEvents,
             enable_worker_overviews: false,
             filter: EventFilter::new(None, flags)
         })),

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

@@ -7,7 +7,9 @@ use crate::rpc_call;
 use crate::server::bootstrap::get_client_session;
 use crate::server::event::journal::JournalReader;
 use crate::server::event::streamer::EventFilter;
-use crate::transfer::messages::{FromClientMessage, StreamEvents, ToClientMessage};
+use crate::transfer::messages::{
+    FromClientMessage, StreamEvents, StreamEventsMode, ToClientMessage,
+};
 use anyhow::anyhow;
 use clap::{Parser, ValueHint};
 use std::io::{BufWriter, Write};
@@ -61,24 +63,19 @@ struct ExportOpts {
 pub async fn command_journal(gsettings: &GlobalSettings, opts: JournalOpts) -> anyhow::Result<()> {
     match opts.command {
         JournalCommand::Export(opts) => export_json(opts),
-        JournalCommand::Replay => stream_json(gsettings, true, false).await,
-        JournalCommand::Stream => stream_json(gsettings, true, true).await,
+        JournalCommand::Replay => stream_json(gsettings, StreamEventsMode::PastEvents).await,
+        JournalCommand::Stream => stream_json(gsettings, StreamEventsMode::PastAndLiveEvents).await,
         JournalCommand::Prune => prune_journal(gsettings).await,
         JournalCommand::Flush => flush_journal(gsettings).await,
     }
 }
 
-async fn stream_json(
-    gsettings: &GlobalSettings,
-    past_events: bool,
-    live_events: bool,
-) -> anyhow::Result<()> {
+async fn stream_json(gsettings: &GlobalSettings, mode: StreamEventsMode) -> anyhow::Result<()> {
     let mut connection = get_client_session(gsettings.server_directory()).await?;
     connection
         .connection()
         .send(FromClientMessage::StreamEvents(StreamEvents {
-            past_events,
-            live_events,
+            mode,
             enable_worker_overviews: false,
             filter: EventFilter::all_events(),
         }))

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

@@ -1,7 +1,7 @@
 use crate::client::output::json::format_datetime;
 
-use crate::server::event::payload::EventPayload;
 use crate::server::event::Event;
+use crate::server::event::payload::EventPayload;
 use crate::transfer::messages::{
     JobSubmitDescription, JobTaskDescription, SubmitRequest, TaskDescription,
 };

crates/hyperqueue/src/client/commands/submit/command.rs

@@ -19,8 +19,8 @@ use crate::server::event::streamer::{EventFilter, EventFilterFlags};
 use crate::transfer::connection::ClientSession;
 use crate::transfer::messages::{
     FromClientMessage, JobDescription, JobSubmitDescription, JobTaskDescription, PinMode,
-    StreamEvents, SubmitRequest, SubmitResponse, TaskDescription, TaskKind, TaskKindProgram,
-    ToClientMessage,
+    StreamEvents, StreamEventsMode, SubmitRequest, SubmitResponse, TaskDescription, TaskKind,
+    TaskKindProgram, ToClientMessage,
 };
 use anyhow::{anyhow, bail};
 use bstr::BString;
@@ -780,8 +780,7 @@ pub(crate) async fn send_submit_request(
             flags.insert(EventFilterFlags::NOTIFY_EVENTS);
         }
         Some(StreamEvents {
-            past_events: false,
-            live_events: true,
+            mode: StreamEventsMode::LiveEvents,
             enable_worker_overviews: false,
             filter: EventFilter::new(None, flags),
         })

crates/hyperqueue/src/common/cli.rs

@@ -400,10 +400,6 @@ pub struct JobWaitOpts {
     /// Waits until all tasks are completed, even if the job is still open
     #[clap(long, action)]
     pub without_close: bool,
-
-    /// Call the program in the argument that is executed whenever a task produce a notification
-    #[clap(long, action)]
-    pub run_on_notify: Option<String>,
 }
 
 #[derive(Parser)]

crates/hyperqueue/src/dashboard/mod.rs

@@ -13,7 +13,9 @@ use crate::server::event::Event;
 use crate::server::event::journal::JournalReader;
 use crate::server::event::streamer::EventFilter;
 use crate::transfer::connection::ClientSession;
-use crate::transfer::messages::{FromClientMessage, StreamEvents, ToClientMessage};
+use crate::transfer::messages::{
+    FromClientMessage, StreamEvents, StreamEventsMode, ToClientMessage,
+};
 use std::time::Duration;
 
 // The time range in which the live timeline is display ([now() - duration, now()])
@@ -47,8 +49,7 @@ pub async fn preload_dashboard_events(
             // Start streaming events
             connection
                 .send(FromClientMessage::StreamEvents(StreamEvents {
-                    past_events: true,
-                    live_events: true,
+                    mode: StreamEventsMode::PastAndLiveEvents,
                     enable_worker_overviews: true,
                     filter: EventFilter::all_events(),
                 }))

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

@@ -139,23 +139,19 @@ async fn start_streaming<
     Tx::Error: Debug,
 {
     let StreamEvents {
-        past_events,
-        live_events,
+        mode,
         enable_worker_overviews,
         filter,
     } = stream_events;
     if enable_worker_overviews {
         senders.server_control.add_worker_overview_listener();
     }
     log::debug!("Start streaming events to client");
-    if !live_events && !past_events {
-        return;
-    }
 
     /* We create two event queues, one for historic events and one for live events
     So while historic events are loaded from the file and streamed, live events are already
     collected and sent immediately once the historic events are sent */
-    let live = if live_events {
+    let live = if mode.is_live_events_enabled() {
         let (tx2, rx2) = mpsc::unbounded_channel::<Event>();
         let listener_id = senders.events.register_listener(filter, tx2);
         Some((rx2, listener_id))
@@ -170,7 +166,7 @@ async fn start_streaming<
     // If we use a journal, we can replay historical events from it.
     // If not, we can at least try to reconstruct a few basic events
     // based on the current state.
-    if past_events {
+    if mode.is_past_events_enabled() {
         let (tx1, rx1) = mpsc::unbounded_channel::<Event>();
         if senders.events.is_journal_enabled() {
             senders.events.start_journal_replay(tx1);
@@ -181,7 +177,7 @@ async fn start_streaming<
     }
 
     if let Some((rx2, listener_id)) = live {
-        if past_events {
+        if mode.is_past_events_enabled() {
             let _ = tx.send(ToClientMessage::EventLiveBoundary).await;
         }
         crate::server::client::stream_events(&mut tx, &mut rx, rx2).await;

crates/hyperqueue/src/transfer/messages.rs

@@ -98,17 +98,39 @@ pub struct TaskKindProgram {
     pub task_dir: bool,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum StreamEventsMode {
+    PastEvents,
+    LiveEvents,
+    PastAndLiveEvents,
+}
+
+impl StreamEventsMode {
+    pub fn is_live_events_enabled(&self) -> bool {
+        matches!(
+            self,
+            StreamEventsMode::LiveEvents | StreamEventsMode::PastAndLiveEvents
+        )
+    }
+
+    pub fn is_past_events_enabled(&self) -> bool {
+        matches!(
+            self,
+            StreamEventsMode::PastEvents | StreamEventsMode::PastAndLiveEvents
+        )
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct StreamEvents {
-    /// If true, replay historical events and then start streaming live events.
-    pub past_events: bool,
-    /// If true, and then start streaming live events.
-    /// If both `live_events` and `past_events` are true, then historical events are replayed first
-    /// then event `EventLiveBoundary` is sent, then live events are streamed.
-    pub live_events: bool,
-    /// Enable worker overviews during the event streaming.
+    /// When the mode is PathAndLiveEvents then:
+    /// 1) historical events are replayed;
+    /// 2) event `EventLiveBoundary` is sent,
+    /// 3) live events are streamed.
+    pub mode: StreamEventsMode,
+    /// Enable worker overviews during the event streaming
     pub enable_worker_overviews: bool,
-    /// Stream events only relevant for a given jobs
+    /// Stream events only relevant for given jobs
     pub filter: EventFilter,
 }