Allow configuring resampler's max_age_in_intervals in LogicalMeter

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>

Author: shsms

src/logical_meter/config.rs

@@ -16,6 +16,9 @@ pub struct LogicalMeterConfig {
     pub(crate) resampling_function: Option<ResamplingFunction<f32, Sample<f32>>>,
     /// Resampler overrides.
     pub(crate) resampling_overrides: HashMap<Metric, ResamplingFunction<f32, Sample<f32>>>,
+    /// The maximum age of samples to be considered for resampling, in number of
+    /// intervals.
+    pub(crate) max_age_in_intervals: u32,
 }
 
 impl LogicalMeterConfig {
@@ -25,6 +28,7 @@ impl LogicalMeterConfig {
             resampling_interval,
             resampling_function: None,
             resampling_overrides: HashMap::new(),
+            max_age_in_intervals: 3,
         }
     }
 
@@ -55,4 +59,17 @@ impl LogicalMeterConfig {
 
         self
     }
+
+    /// Sets the maximum age of samples to be considered for resampling, in
+    /// number of intervals.
+    ///
+    /// Must be at least 1.  If a smaller value is provided, it will be clamped
+    /// to 1.
+    ///
+    /// If not set, the default value is 3.
+    pub fn with_max_age_in_intervals(mut self, max_age_in_intervals: u32) -> Self {
+        // Ensure that the maximum age is at least 1 interval.
+        self.max_age_in_intervals = max_age_in_intervals.max(1);
+        self
+    }
 }

src/logical_meter/logical_meter_actor.rs

@@ -383,7 +383,9 @@ impl LogicalMeterActor {
                         // Finally, default to average if no default is
                         // configured
                         .unwrap_or(ResamplingFunction::Average),
-                    3,
+                    // The resampler expects max age to be i32, so we need to
+                    // cap it if the user provided a higher value.
+                    self.config.max_age_in_intervals.min(i32::MAX as u32) as i32,
                     self.resampler_ts,
                     false,
                 ),

src/logical_meter/logical_meter_handle.rs

@@ -473,6 +473,61 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn test_max_age_in_intervals() {
+        let lm_config = Some(
+            LogicalMeterConfig::new(TimeDelta::try_milliseconds(200).unwrap())
+                .with_max_age_in_intervals(1)
+                .with_default_resampling_function(ResamplingFunction::Count),
+        );
+        let mut lm = new_logical_meter_handle(lm_config).await;
+        let formula = lm.consumer(crate::metric::AcPowerActive).unwrap();
+
+        let samples = fetch_samples(formula, 8).await;
+        check_samples(
+            samples,
+            |q| q.as_watts(),
+            TimeDelta::try_milliseconds(200).unwrap(),
+            vec![
+                Some(1.0),
+                Some(1.0),
+                Some(1.0),
+                Some(1.0),
+                Some(1.0),
+                Some(1.0),
+                Some(0.0),
+                Some(0.0),
+            ],
+        );
+
+        let lm_config = Some(
+            LogicalMeterConfig::new(TimeDelta::try_milliseconds(200).unwrap())
+                .with_max_age_in_intervals(3)
+                .with_default_resampling_function(ResamplingFunction::Count),
+        );
+        let mut lm = new_logical_meter_handle(lm_config).await;
+        let formula = lm.consumer(crate::metric::AcPowerActive).unwrap();
+
+        let samples = fetch_samples(formula, 10).await;
+        check_samples(
+            samples,
+            |q| q.as_watts(),
+            TimeDelta::try_milliseconds(200).unwrap(),
+            vec![
+                Some(1.0),
+                Some(2.0),
+                Some(3.0),
+                Some(3.0),
+                Some(3.0),
+                Some(3.0),
+                Some(2.0),
+                Some(1.0),
+                Some(0.0),
+                Some(0.0),
+            ],
+        )
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_consumer_current_formula() {
         let formula = new_logical_meter_handle(None)