Adds aggregations example application

SirCipher · SirCipher · commit a34e203a840f · 2024-06-14T18:48:21.000+01:00
diff --git a/example_apps/aggregations/src/aggregate.rs b/example_apps/aggregations/src/aggregate.rs
@@ -0,0 +1,49 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use swimos::{
+    agent::agent_lifecycle::utility::HandlerContext,
+    agent::event_handler::EventHandler,
+    agent::lanes::{CommandLane, JoinValueLane},
+    agent::AgentLaneModel,
+    agent::{lifecycle, projections},
+};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct AggregateAgent {
+    aggregated: JoinValueLane<String, f64>,
+    register: CommandLane<String>,
+}
+
+#[derive(Clone, Default)]
+pub struct AggregateLifecycle;
+
+#[lifecycle(AggregateAgent)]
+impl AggregateLifecycle {
+    #[on_command(register)]
+    pub fn register(
+        &self,
+        context: HandlerContext<AggregateAgent>,
+        area_id: &String,
+    ) -> impl EventHandler<AggregateAgent> {
+        context.add_downlink(
+            AggregateAgent::AGGREGATED,
+            area_id.clone(),
+            None,
+            format!("/area/{}", area_id).as_str(),
+            "average_speed",
+        )
+    }
+}
diff --git a/example_apps/aggregations/src/area.rs b/example_apps/aggregations/src/area.rs
@@ -1,8 +1,27 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::sync::{Arc, Mutex};
+
 use rand::seq::SliceRandom;
 
 use swimos::{
     agent::agent_lifecycle::utility::HandlerContext,
     agent::event_handler::EventHandler,
+    agent::event_handler::HandlerActionExt,
     agent::lanes::{CommandLane, JoinValueLane, ValueLane},
     agent::lifecycle,
     agent::projections,
@@ -17,6 +36,20 @@ pub enum Area {
     D,
 }
 
+impl FromStr for Area {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "A" => Ok(Area::A),
+            "B" => Ok(Area::B),
+            "C" => Ok(Area::C),
+            "D" => Ok(Area::D),
+            _ => Err(()),
+        }
+    }
+}
+
 impl Area {
     pub fn random() -> Self {
         [Area::A, Area::B, Area::C, Area::D]
@@ -36,23 +69,42 @@ pub struct AreaAgent {
     average_speed: ValueLane<f64>,
 }
 
-#[derive(Clone)]
-pub struct AreaLifecycle;
+#[derive(Clone, Default)]
+pub struct AreaLifecycle {
+    area: Arc<Mutex<Option<Area>>>,
+}
 
 #[lifecycle(AreaAgent)]
 impl AreaLifecycle {
+    #[on_start]
+    pub fn on_start(&self, context: HandlerContext<AreaAgent>) -> impl EventHandler<AreaAgent> {
+        let area = self.area.clone();
+        context
+            .get_parameter("area")
+            .and_then(move |area_id: Option<String>| {
+                context.effect(move || {
+                    let assigned_area = &mut *area.lock().expect("Mutex poisoned");
+                    let area_str = area_id.expect("Missing area URI");
+                    *assigned_area = Area::from_str(area_str.as_str()).ok();
+                    area_str
+                })
+            })
+            .and_then(move |area: String| {
+                context.send_command(None, "/aggregate", "register", area)
+            })
+    }
+
     #[on_command(register)]
     pub fn register(
         &self,
         context: HandlerContext<AreaAgent>,
         car_id: &u64,
     ) -> impl EventHandler<AreaAgent> {
-        let car_id = *car_id;
         context.add_downlink(
-            AreaAgent::REGISTER,
-            car_id,
+            AreaAgent::CARS,
+            *car_id,
             None,
-            car_id.to_string().as_str(),
+            format!("/cars/{car_id}").as_str(),
             "speed",
         )
     }
@@ -61,8 +113,23 @@ impl AreaLifecycle {
     pub fn deregister(
         &self,
         context: HandlerContext<AreaAgent>,
-        _car_id: &u64,
+        car_id: &u64,
     ) -> impl EventHandler<AreaAgent> {
+        context.remove_downlink(AreaAgent::CARS, *car_id)
+    }
 
+    #[on_update(cars)]
+    fn cars(
+        &self,
+        context: HandlerContext<AreaAgent>,
+        speeds: &HashMap<u64, u64>,
+        _key: u64,
+        _prev: Option<u64>,
+        _new_value: &u64,
+    ) -> impl EventHandler<AreaAgent> {
+        let speeds = speeds.clone();
+        context
+            .effect(move || speeds.values().sum::<u64>() as f64 / speeds.len() as f64)
+            .and_then(move |average: f64| context.set_value(AreaAgent::AVERAGE_SPEED, average))
     }
 }
diff --git a/example_apps/aggregations/src/car.rs b/example_apps/aggregations/src/car.rs
@@ -1,3 +1,17 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 use std::mem::replace;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
@@ -39,13 +53,12 @@ impl CarLifecycle {
     pub fn on_start(&self, context: HandlerContext<CarAgent>) -> impl EventHandler<CarAgent> {
         let area = self.area.clone();
 
-        let speed_handler = context.schedule_repeatedly(Duration::from_secs(10), move || {
+        let speed_handler = context.schedule_repeatedly(Duration::from_secs(5), move || {
             let mut rng = rand::rngs::OsRng::default();
             Some(context.set_value(CarAgent::SPEED, rng.gen_range(10..=30)))
         });
-
         let car_handler = move |car_id: u64| {
-            context.schedule_repeatedly(Duration::from_secs(5), move || {
+            context.schedule_repeatedly(Duration::from_secs(1), move || {
                 let area = area.clone();
                 let car_id = car_id.clone();
 
@@ -54,21 +67,21 @@ impl CarLifecycle {
 
                 let handler = if old_area != *assigned_area {
                     // deregister this car with its current area
-                    let register_handler = context.send_command(
+                    let deregister_handler = context.send_command(
                         None,
                         format!("/area/{old_area:?}"),
                         "deregister".to_string(),
                         car_id.clone(),
                     );
                     // register this car with its new assigned area
-                    let deregister_handler = context.send_command(
+                    let register_handler = context.send_command(
                         None,
                         format!("/area/{:?}", *assigned_area),
                         "register".to_string(),
                         car_id,
                     );
 
-                    Some(register_handler.followed_by(deregister_handler))
+                    Some(deregister_handler.followed_by(register_handler))
                 } else {
                     // noop handler as the car didn't switch area
                     None
diff --git a/example_apps/aggregations/src/main.rs b/example_apps/aggregations/src/main.rs
@@ -1,15 +1,36 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 use std::error::Error;
+use std::str::FromStr;
 use std::time::Duration;
 
+use crate::{
+    aggregate::{AggregateAgent, AggregateLifecycle},
+    area::{AreaAgent, AreaLifecycle},
+    car::CarAgent,
+    car::CarLifecycle,
+};
 use example_util::{example_logging, manage_handle};
+use swimos::route::RouteUri;
 use swimos::{
     agent::agent_model::AgentModel,
     route::RoutePattern,
     server::{Server, ServerBuilder},
 };
 
-use crate::car::{CarAgent, CarLifecycle};
-
+mod aggregate;
 mod area;
 mod car;
 
@@ -18,26 +39,38 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     example_logging()?;
 
     let car_agent = AgentModel::new(CarAgent::default, CarLifecycle::default().into_lifecycle());
+    let area_agent = AgentModel::new(
+        AreaAgent::default,
+        AreaLifecycle::default().into_lifecycle(),
+    );
+    let aggregate_agent = AgentModel::new(
+        AggregateAgent::default,
+        AggregateLifecycle::default().into_lifecycle(),
+    );
 
     let server = ServerBuilder::with_plane_name("Example Plane")
-        .add_route(RoutePattern::parse_str("/car/:car_id")?, car_agent)
+        .add_route(RoutePattern::parse_str("/cars/:car_id")?, car_agent)
+        .add_route(RoutePattern::parse_str("/area/:area")?, area_agent)
+        .add_route(RoutePattern::parse_str("/aggregate")?, aggregate_agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
         })
         .build()
         .await?;
 
     let (task, handle) = server.run();
-    // handle
-    //     .start_agent(RouteUri::from_str("/generator")?)
-    //     .await
-    //     .expect("Failed to start generator agent");
+    let _task = tokio::spawn(task);
 
-    let shutdown = manage_handle(handle);
+    for i in 0..1000 {
+        let route = format!("/cars/{i}");
+        handle
+            .start_agent(RouteUri::from_str(route.as_str())?)
+            .await
+            .expect("Failed to start agent");
+    }
 
-    let (_, result) = tokio::join!(shutdown, task);
-
-    result?;
+    manage_handle(handle).await;
     println!("Server stopped successfully.");
+
     Ok(())
 }
