Merge pull request #655 from swimos/aggregations

SirCipher · web-flow · commit 20202c3483e5 · 2024-06-19T09:18:21.000+01:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -38,6 +38,7 @@ members = [
     "example_apps/tutorial_app/generator",
     "example_apps/join_map",
     "example_apps/join_value",
+    "example_apps/aggregations",
 ]
 
 exclude = [
@@ -126,4 +127,4 @@ httparse = "1.8"
 sha1 = "0.10"
 waker-fn = "1.1.0"
 num = "0.4"
-smol_str = "0.2.0"
+smol_str = "0.2.0"
diff --git a/example_apps/aggregations/Cargo.toml b/example_apps/aggregations/Cargo.toml
@@ -0,0 +1,12 @@
+[package]
+name = "aggregations"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+swimos = { path = "../../swimos", features = ["server", "agent"] }
+swimos_form = { path = "../../api/swimos_form" }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+example-util = { path = "../example_util" }
+rand = { workspace = true }
+futures-util = { workspace = true }
diff --git a/example_apps/aggregations/src/aggregate.rs b/example_apps/aggregations/src/aggregate.rs
@@ -0,0 +1,65 @@
+// 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 swimos::{
+    agent::agent_lifecycle::utility::HandlerContext,
+    agent::event_handler::EventHandler,
+    agent::lanes::ValueLane,
+    agent::lanes::{CommandLane, JoinValueLane},
+    agent::AgentLaneModel,
+    agent::{lifecycle, projections},
+};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct AggregateAgent {
+    aggregated: JoinValueLane<String, f64>,
+    average_speed: ValueLane<f64>,
+    register: CommandLane<String>,
+}
+
+#[derive(Clone, Default)]
+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 average = averages.values().sum::<f64>() / averages.len() as f64;
+        context.set_value(AggregateAgent::AVERAGE_SPEED, average)
+    }
+
+    #[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
@@ -0,0 +1,159 @@
+// 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::fmt::{Display, Formatter};
+use std::str::FromStr;
+
+use rand::seq::SliceRandom;
+
+use swimos::agent::event_handler::HandlerActionExt;
+use swimos::{
+    agent::agent_lifecycle::utility::HandlerContext,
+    agent::event_handler::EventHandler,
+    agent::lanes::{CommandLane, JoinValueLane, ValueLane},
+    agent::lifecycle,
+    agent::projections,
+    agent::AgentLaneModel,
+};
+use swimos_form::Form;
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Form)]
+pub enum Area {
+    Arbury,
+    CherryHinton,
+    KingsHedges,
+    Petersfield,
+}
+
+impl Area {
+    pub fn universe() -> [Area; 4] {
+        [
+            Area::Arbury,
+            Area::CherryHinton,
+            Area::KingsHedges,
+            Area::Petersfield,
+        ]
+    }
+}
+
+impl Default for Area {
+    fn default() -> Self {
+        Area::select_random()
+    }
+}
+
+impl FromStr for Area {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "arbury" => Ok(Area::Arbury),
+            "cherryhinton" => Ok(Area::CherryHinton),
+            "kingshedges" => Ok(Area::KingsHedges),
+            "petersfield" => Ok(Area::Petersfield),
+            _ => Err(()),
+        }
+    }
+}
+
+impl Display for Area {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Area::Arbury => write!(f, "arbury"),
+            Area::CherryHinton => write!(f, "cherryhinton"),
+            Area::KingsHedges => write!(f, "kingshedges"),
+            Area::Petersfield => write!(f, "petersfield"),
+        }
+    }
+}
+
+impl Area {
+    pub fn select_random() -> Self {
+        [
+            Area::Arbury,
+            Area::CherryHinton,
+            Area::KingsHedges,
+            Area::Petersfield,
+        ]
+        .choose(&mut rand::thread_rng())
+        .copied()
+        .expect("Slice was not empty")
+    }
+}
+
+#[derive(PartialEq, Copy, Clone, Form)]
+pub enum Action {
+    Register(u64),
+    Deregister(u64),
+}
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct AreaAgent {
+    registrations: CommandLane<Action>,
+    cars: JoinValueLane<u64, u64>,
+    average_speed: ValueLane<f64>,
+}
+
+#[derive(Clone, Default)]
+pub struct AreaLifecycle {
+    area: Area,
+}
+
+impl AreaLifecycle {
+    pub fn new(area: Area) -> AreaLifecycle {
+        AreaLifecycle { area }
+    }
+}
+
+#[lifecycle(AreaAgent)]
+impl AreaLifecycle {
+    #[on_start]
+    pub fn on_start(&self, context: HandlerContext<AreaAgent>) -> impl EventHandler<AreaAgent> {
+        context.send_command(None, "/aggregate", "register", self.area.to_string())
+    }
+
+    #[on_command(registrations)]
+    pub fn registrations(
+        &self,
+        context: HandlerContext<AreaAgent>,
+        action: &Action,
+    ) -> impl EventHandler<AreaAgent> {
+        match action {
+            Action::Register(car_id) => context
+                .add_downlink(
+                    AreaAgent::CARS,
+                    *car_id,
+                    None,
+                    format!("/cars/{car_id}").as_str(),
+                    "speed",
+                )
+                .boxed(),
+            Action::Deregister(car_id) => context.remove_downlink(AreaAgent::CARS, *car_id).boxed(),
+        }
+    }
+    #[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 average = speeds.values().sum::<u64>() as f64 / speeds.len() as f64;
+        context.set_value(AreaAgent::AVERAGE_SPEED, average)
+    }
+}
diff --git a/example_apps/aggregations/src/car.rs b/example_apps/aggregations/src/car.rs
@@ -0,0 +1,92 @@
+// 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::str::FromStr;
+use std::time::Duration;
+
+use rand::Rng;
+
+use swimos::agent::stores::ValueStore;
+use swimos::{
+    agent::agent_lifecycle::utility::HandlerContext,
+    agent::event_handler::{EventHandler, HandlerActionExt},
+    agent::lanes::ValueLane,
+    agent::{lifecycle, projections, AgentLaneModel},
+};
+
+use crate::area::{Action, Area};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct CarAgent {
+    speed: ValueLane<u64>,
+    #[item(transient)]
+    area: ValueStore<Area>,
+}
+
+#[derive(Debug, Clone)]
+pub struct CarLifecycle;
+
+#[lifecycle(CarAgent)]
+impl CarLifecycle {
+    #[on_start]
+    pub fn on_start(&self, context: HandlerContext<CarAgent>) -> impl EventHandler<CarAgent> {
+        let speed_handler = context.schedule_repeatedly(Duration::from_secs(5), move || {
+            let mut rng = rand::rngs::OsRng;
+            Some(context.set_value(CarAgent::SPEED, rng.gen_range(10..=70)))
+        });
+        let area_handler = move |car_id: u64| {
+            context.schedule_repeatedly(Duration::from_secs(5), move || {
+                Some(context.get_value(CarAgent::AREA).and_then(move |old_area| {
+                    let new_area = Area::select_random();
+                    let handler = if old_area != new_area {
+                        // deregister this car with its current area
+                        let deregister_handler = context.send_command(
+                            None,
+                            format!("/area/{old_area}"),
+                            "registrations".to_string(),
+                            Action::Deregister(car_id),
+                        );
+                        // register this car with its new assigned area
+                        let register_handler = context.send_command(
+                            None,
+                            format!("/area/{new_area}"),
+                            "registrations".to_string(),
+                            Action::Register(car_id),
+                        );
+
+                        let handler = deregister_handler
+                            .followed_by(register_handler)
+                            .followed_by(context.set_value(CarAgent::AREA, new_area));
+                        Some(handler)
+                    } else {
+                        // noop handler as the car didn't switch area
+                        None
+                    };
+
+                    handler.discard()
+                }))
+            })
+        };
+
+        context
+            .get_parameter("car_id")
+            .map(|param: Option<String>| {
+                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(area_handler)
+            .followed_by(speed_handler)
+    }
+}
diff --git a/example_apps/aggregations/src/main.rs b/example_apps/aggregations/src/main.rs
