Fix CI clippy warnings

Author: argahsuknesib

src/paper_bench/harness.rs

@@ -2354,7 +2354,6 @@ fn run_sustained_system(
 
     let live_start_at;
     let mut external_join_latencies = Vec::<(i64, f64)>::new();
-    let mut live_rows_count = 0;
     let historical_baseline;
 
     let estimated_external_transfer_bytes_total;
@@ -2496,8 +2495,6 @@ fn run_sustained_system(
                         if win_end <= horizon_end_ms {
                             all_live_rows.push(rows.clone());
                         }
-                        live_rows_count += 1;
-
                         let start_join = Instant::now();
                         let joined = join_live_with_baseline(&[rows], &materialized_baseline_rows);
                         let join_duration = start_join.elapsed().as_secs_f64() * 1000.0;
@@ -2545,9 +2542,7 @@ fn run_sustained_system(
                     if win_end <= horizon_end_ms {
                         all_live_rows.push(rows.clone());
                     }
-                    live_rows_count += 1;
-
-                    let start_join = Instant::now();
+                        let start_join = Instant::now();
                     let joined = join_live_with_baseline(&[rows], &materialized_baseline_rows);
                     let join_duration = start_join.elapsed().as_secs_f64() * 1000.0;
                     if first_hybrid_result_at == 0 {
@@ -2636,7 +2631,9 @@ fn run_sustained_system(
     };
 
     let estimated_external_transfer_bytes_per_window = if completed_windows_in_horizon > 0 {
-        estimated_external_transfer_bytes_total / completed_windows_in_horizon
+        estimated_external_transfer_bytes_total
+            .checked_div(completed_windows_in_horizon)
+            .unwrap_or(0)
     } else {
         0
     };

src/paper_bench/query_defined_baseline.rs

@@ -1447,7 +1447,7 @@ fn collect_live_results(
     first_event_started: Instant,
     expected_results: usize,
 ) -> Result<Vec<TimedBinding>, Box<dyn std::error::Error>> {
-    let deadline = Instant::now() + Duration::from_secs(600);
+    let deadline = Instant::now() + Duration::from_secs(10 * 60);
     let mut timed_results = Vec::new();
     let mut last_progress = Instant::now();
 
@@ -1485,8 +1485,8 @@ fn collect_live_results(
 fn expected_live_averages(live_events: &[RDFEvent]) -> HashMap<String, f64> {
     let mut by_sensor: HashMap<String, (f64, usize)> = HashMap::new();
     for event in live_events {
-        let sensor = normalize_binding_term(&event.subject.to_string());
-        let value = event.object.to_string().trim().parse::<f64>().unwrap_or(0.0);
+        let sensor = normalize_binding_term(&event.subject.clone());
+        let value = event.object.clone().trim().parse::<f64>().unwrap_or(0.0);
         let entry = by_sensor.entry(sensor).or_insert((0.0, 0));
         entry.0 += value;
         entry.1 += 1;

src/stream_bus/stream_bus.rs

@@ -308,10 +308,14 @@ impl StreamBus {
         F: Fn(RDFEvent, String) -> Fut + Send + Sync,
         Fut: std::future::Future<Output = ()> + Send,
     {
-        let interval = if self.config.rate_of_publishing > 0 {
-            Some(Duration::from_micros(1_000_000 / self.config.rate_of_publishing))
-        } else {
+        let interval = if self.config.rate_of_publishing == 0 {
             None
+        } else {
+            Some(Duration::from_micros(
+                1_000_000u64
+                    .checked_div(self.config.rate_of_publishing)
+                    .expect("rate_of_publishing is positive"),
+            ))
         };
 
         let mut last_report = Instant::now();