Tidy up

Author: SirCipher

example_apps/aggregations/src/aggregate.rs

@@ -12,9 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use swimos::{
     agent::agent_lifecycle::utility::HandlerContext,
     agent::event_handler::EventHandler,
+    agent::event_handler::HandlerActionExt,
+    agent::lanes::ValueLane,
     agent::lanes::{CommandLane, JoinValueLane},
     agent::AgentLaneModel,
     agent::{lifecycle, projections},
@@ -24,6 +27,7 @@ use swimos::{
 #[projections]
 pub struct AggregateAgent {
     aggregated: JoinValueLane<String, f64>,
+    average_speed: ValueLane<f64>,
     register: CommandLane<String>,
 }
 
@@ -32,6 +36,21 @@ pub struct AggregateLifecycle;
 
 #[lifecycle(AggregateAgent)]
 impl AggregateLifecycle {
+    #[on_update(aggregated)]
+    fn aggregated(
+        &self,
+        context: HandlerContext<AggregateAgent>,
+        averages: &HashMap<String, f64>,
+        _key: String,
+        _prev: Option<f64>,
+        _new_value: &f64,
+    ) -> impl EventHandler<AggregateAgent> {
+        let averages = averages.clone();
+        context
+            .effect(move || averages.values().sum::<f64>() / averages.len() as f64)
+            .and_then(move |average: f64| context.set_value(AggregateAgent::AVERAGE_SPEED, average))
+    }
+
     #[on_command(register)]
     pub fn register(
         &self,

example_apps/aggregations/src/area.rs

@@ -64,7 +64,6 @@ impl Area {
 pub struct AreaAgent {
     register: CommandLane<u64>,
     deregister: CommandLane<u64>,
-    // car_id -> speed
     cars: JoinValueLane<u64, u64>,
     average_speed: ValueLane<f64>,
 }

example_apps/aggregations/src/car.rs

@@ -54,14 +54,12 @@ impl CarLifecycle {
         let area = self.area.clone();
 
         let speed_handler = context.schedule_repeatedly(Duration::from_secs(5), move || {
-            let mut rng = rand::rngs::OsRng::default();
+            let mut rng = rand::rngs::OsRng;
             Some(context.set_value(CarAgent::SPEED, rng.gen_range(10..=70)))
         });
-        let car_handler = move |car_id: u64| {
-            context.schedule_repeatedly(Duration::from_secs(1), move || {
+        let area_handler = move |car_id: u64| {
+            context.schedule_repeatedly(Duration::from_secs(5), move || {
                 let area = area.clone();
-                let car_id = car_id.clone();
-
                 let assigned_area = &mut *area.lock().expect("Mutex poisoned");
                 let old_area = replace(assigned_area, Area::random());
 
@@ -71,7 +69,7 @@ impl CarLifecycle {
                         None,
                         format!("/area/{old_area:?}"),
                         "deregister".to_string(),
-                        car_id.clone(),
+                        car_id,
                     );
                     // register this car with its new assigned area
                     let register_handler = context.send_command(
@@ -97,7 +95,7 @@ impl CarLifecycle {
                 let car_id = param.expect("Missing car_id URI parameter");
                 u64::from_str(car_id.as_str()).expect("Failed to parse car ID into u64")
             })
-            .and_then(car_handler)
+            .and_then(area_handler)
             .followed_by(speed_handler)
     }
 }

example_apps/aggregations/src/main.rs

@@ -43,10 +43,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         AreaAgent::default,
         AreaLifecycle::default().into_lifecycle(),
     );
-    let aggregate_agent = AgentModel::new(
-        AggregateAgent::default,
-        AggregateLifecycle::default().into_lifecycle(),
-    );
+    let aggregate_agent =
+        AgentModel::new(AggregateAgent::default, AggregateLifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
         .add_route(RoutePattern::parse_str("/cars/:car_id")?, car_agent)