Resolves PR comments

Author: SirCipher

example_apps/aggregations/Cargo.toml

@@ -5,6 +5,7 @@ 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 }

example_apps/aggregations/src/aggregate.rs

@@ -16,7 +16,6 @@ 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,
@@ -45,10 +44,8 @@ impl AggregateLifecycle {
         _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))
+        let average = averages.values().sum::<f64>() / averages.len() as f64;
+        context.set_value(AggregateAgent::AVERAGE_SPEED, average)
     }
 
     #[on_command(register)]

example_apps/aggregations/src/area.rs

@@ -13,110 +13,137 @@
 // limitations under the License.
 
 use std::collections::HashMap;
+use std::fmt::{Display, Formatter};
 use std::str::FromStr;
-use std::sync::{Arc, Mutex};
 
 use rand::seq::SliceRandom;
 
+use swimos::agent::event_handler::HandlerActionExt;
 use swimos::{
     agent::agent_lifecycle::utility::HandlerContext,
     agent::event_handler::EventHandler,
-    agent::event_handler::HandlerActionExt,
     agent::lanes::{CommandLane, JoinValueLane, ValueLane},
     agent::lifecycle,
     agent::projections,
     agent::AgentLaneModel,
 };
+use swimos_form::Form;
 
-#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Form)]
 pub enum Area {
-    A,
-    B,
-    C,
-    D,
+    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 {
-            "A" => Ok(Area::A),
-            "B" => Ok(Area::B),
-            "C" => Ok(Area::C),
-            "D" => Ok(Area::D),
+            "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 random() -> Self {
-        [Area::A, Area::B, Area::C, Area::D]
-            .choose(&mut rand::thread_rng())
-            .copied()
-            .expect("Slice was not empty")
+    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 {
-    register: CommandLane<u64>,
-    deregister: CommandLane<u64>,
+    registrations: CommandLane<Action>,
     cars: JoinValueLane<u64, u64>,
     average_speed: ValueLane<f64>,
 }
 
 #[derive(Clone, Default)]
 pub struct AreaLifecycle {
-    area: Arc<Mutex<Option<Area>>>,
+    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> {
-        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> {
-        context.add_downlink(
-            AreaAgent::CARS,
-            *car_id,
-            None,
-            format!("/cars/{car_id}").as_str(),
-            "speed",
-        )
+        context.send_command(None, "/aggregate", "register", self.area.to_string())
     }
 
-    #[on_command(deregister)]
-    pub fn deregister(
+    #[on_command(registrations)]
+    pub fn registrations(
         &self,
         context: HandlerContext<AreaAgent>,
-        car_id: &u64,
+        action: &Action,
     ) -> impl EventHandler<AreaAgent> {
-        context.remove_downlink(AreaAgent::CARS, *car_id)
+        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,
@@ -126,9 +153,7 @@ impl AreaLifecycle {
         _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))
+        let average = speeds.values().sum::<u64>() as f64 / speeds.len() as f64;
+        context.set_value(AreaAgent::AVERAGE_SPEED, average)
     }
 }

example_apps/aggregations/src/car.rs

@@ -12,80 +12,71 @@
 // 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};
 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::Area;
+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 {
-    area: Arc<Mutex<Area>>,
-}
-
-impl Default for CarLifecycle {
-    fn default() -> Self {
-        CarLifecycle {
-            area: Arc::new(Mutex::new(Area::random())),
-        }
-    }
-}
+pub struct CarLifecycle;
 
 #[lifecycle(CarAgent)]
 impl CarLifecycle {
     #[on_start]
     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(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 || {
-                let area = area.clone();
-                let assigned_area = &mut *area.lock().expect("Mutex poisoned");
-                let old_area = replace(assigned_area, Area::random());
-
-                let handler = if old_area != *assigned_area {
-                    // deregister this car with its current area
-                    let deregister_handler = context.send_command(
-                        None,
-                        format!("/area/{old_area:?}"),
-                        "deregister".to_string(),
-                        car_id,
-                    );
-                    // register this car with its new assigned area
-                    let register_handler = context.send_command(
-                        None,
-                        format!("/area/{:?}", *assigned_area),
-                        "register".to_string(),
-                        car_id,
-                    );
+                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),
+                        );
 
-                    Some(deregister_handler.followed_by(register_handler))
-                } else {
-                    // noop handler as the car didn't switch area
-                    None
-                };
+                        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
+                    };
 
-                Some(handler.discard())
+                    handler.discard()
+                }))
             })
         };
 

example_apps/aggregations/src/main.rs

@@ -16,6 +16,7 @@ use std::error::Error;
 use std::str::FromStr;
 use std::time::Duration;
 
+use crate::area::Area;
 use crate::{
     aggregate::{AggregateAgent, AggregateLifecycle},
     area::{AreaAgent, AreaLifecycle},
@@ -38,23 +39,31 @@ mod car;
 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 car_agent = AgentModel::new(CarAgent::default, CarLifecycle.into_lifecycle());
+    let area_agent = move |name| {
+        AgentModel::new(
+            AreaAgent::default,
+            AreaLifecycle::new(name).into_lifecycle(),
+        )
+    };
     let aggregate_agent =
         AgentModel::new(AggregateAgent::default, AggregateLifecycle.into_lifecycle());
 
-    let server = ServerBuilder::with_plane_name("Example Plane")
+    let mut builder = ServerBuilder::with_plane_name("Example Plane")
         .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?;
+        });
+
+    for area in Area::universe() {
+        builder = builder.add_route(
+            RoutePattern::parse_str(format!("/area/{}", area).as_str())?,
+            area_agent(area),
+        );
+    }
+
+    let server = builder.build().await?;
 
     let (task, handle) = server.run();
     let _task = tokio::spawn(task);