Add stop time to howl

src/howl.rs

@@ -30,6 +30,7 @@ fn generate_run_start<'a>(
     det_max: i32,
     event_topic: &str,
     job_id: &str,
+    stop_time: Option<u64>,
 ) -> &'a [u8] {
     fbb.reset();
     let args = SpectraDetectorMappingArgs {
@@ -76,7 +77,7 @@ fn generate_run_start<'a>(
 
     let run_start_args = RunStartArgs {
         start_time: start_time as u64,
-        stop_time: 0, // TODO check this - it's optional so not necessarily 0
+        stop_time: stop_time.unwrap_or_default(),
         run_name: Some(fbb.create_string(&run_name)),
         instrument_name: Some(fbb.create_string("saluki-howl")),
         nexus_structure: Some(fbb.create_string(&nexus_structure.to_string())),
@@ -192,6 +193,7 @@ fn produce_messages(
                     conf.event_message_config.det_max,
                     conf.event_topic,
                     current_job_id,
+                    conf.stop_time,
                 ))
                 .timestamp(now_nanos / 1_000_000),
         ) {
@@ -271,6 +273,7 @@ pub struct HowlConfig<'a> {
     pub messages_per_frame: u32,
     pub frames_per_second: u32,
     pub frames_per_run: u32,
+    pub stop_time: Option<u64>,
     pub veto_probability: f64, // 1 = always vetoed, 0 = never vetoed
     pub event_message_config: &'a EventMessageConfig,
     pub kafka_config: Option<Vec<KafkaOption>>,
@@ -338,6 +341,7 @@ pub fn howl(conf: &HowlConfig) {
                     conf.event_message_config.det_max,
                     conf.event_topic,
                     &current_job_id,
+                    conf.stop_time,
                 ))
                 .timestamp(now_nanos / 1_000_000),
         )
@@ -385,3 +389,21 @@ pub fn howl(conf: &HowlConfig) {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use isis_streaming_data_types::flatbuffers_generated::run_start_pl72::root_as_run_start;
+
+    fn serialised_stop_time(stop_time: Option<u64>) -> u64 {
+        let mut fbb = FlatBufferBuilder::new();
+        let data = generate_run_start(&mut fbb, 2, "events", "job", stop_time);
+        root_as_run_start(data).unwrap().stop_time()
+    }
+
+    #[test]
+    fn run_start_uses_configured_stop_time() {
+        assert_eq!(serialised_stop_time(None), 0);
+        assert_eq!(serialised_stop_time(Some(123456)), 123456);
+    }
+}

src/main.rs

@@ -83,6 +83,8 @@ enum Commands {
         /// Frames to take before beginning new run (0 to run forever)
         #[arg(long, default_value = "0")]
         frames_per_run: u32,
+        #[arg(long, help = "Run stop time in milliseconds since Unix epoch")]
+        stop_time: Option<u64>,
         /// Time-of-flight peak (ns)
         #[arg(long, default_value = "10000000.0")]
         tof_peak: f32,
@@ -116,6 +118,19 @@ enum Commands {
     },
 }
 
+fn validated_stop_time(
+    frames_per_run: u32,
+    stop_time: Option<u64>,
+) -> Result<Option<u64>, clap::Error> {
+    if stop_time.is_some() && frames_per_run > 0 {
+        return Err(clap::Error::raw(
+            clap::error::ErrorKind::ArgumentConflict,
+            "--stop-time requires --frames-per-run to be 0",
+        ));
+    }
+    Ok(stop_time)
+}
+
 #[tokio::main]
 async fn main() {
     let cli = Cli::parse();
@@ -159,6 +174,7 @@ async fn main() {
             messages_per_frame,
             frames_per_second,
             frames_per_run,
+            stop_time,
             tof_peak,
             tof_sigma,
             det_min,
@@ -173,6 +189,8 @@ async fn main() {
             messages_per_frame,
             frames_per_second,
             frames_per_run,
+            stop_time: validated_stop_time(frames_per_run, stop_time)
+                .unwrap_or_else(|error| error.exit()),
             event_message_config: &EventMessageConfig {
                 events_per_message,
                 tof_peak,
@@ -191,3 +209,44 @@ async fn main() {
         } // Commands::Play {} => {}
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::error::ErrorKind;
+
+    #[test]
+    fn howl_stop_time_uses_single_run_default() {
+        let cli = Cli::try_parse_from([
+            "saluki",
+            "howl",
+            "localhost:9092",
+            "TEST",
+            "--stop-time",
+            "123456",
+        ])
+        .unwrap();
+
+        match cli.command {
+            Commands::Howl {
+                frames_per_run,
+                stop_time,
+                ..
+            } => {
+                assert_eq!(frames_per_run, 0);
+                assert_eq!(
+                    validated_stop_time(frames_per_run, stop_time).unwrap(),
+                    Some(123456)
+                );
+            }
+            _ => panic!("expected howl command"),
+        }
+    }
+
+    #[test]
+    fn howl_stop_time_conflicts_with_repeating_runs() {
+        let error = validated_stop_time(10, Some(123456)).unwrap_err();
+
+        assert_eq!(error.kind(), ErrorKind::ArgumentConflict);
+    }
+}