Merge branch 'main' into docs2

Author: horned-sphere

Cargo.toml

@@ -37,6 +37,7 @@ members = [
     "example_apps/tutorial_app",
     "example_apps/tutorial_app/model",
     "example_apps/tutorial_app/generator",
+    "example_apps/join_value",
 ]
 
 exclude = [

api/swimos_agent_protocol/src/ad_hoc/mod.rs

@@ -29,7 +29,7 @@ use crate::AdHocCommand;
 mod tests;
 
 #[derive(Debug, Default, Clone, Copy)]
-pub struct CommandEncoder<E> {
+struct CommandEncoder<E> {
     body_encoder: E,
 }
 

api/swimos_agent_protocol/src/lib.rs

@@ -87,7 +87,7 @@ pub mod encoding {
         };
     }
 
-    /// # The encoding used for map like lanes and stores, shared between the other protocols in this module.
+    /// The encoding used for map like lanes and stores, shared between the other protocols in this module.
     pub mod map {
         pub use crate::map::{
             MapMessageDecoder, MapMessageEncoder, MapOperationDecoder, MapOperationEncoder,
@@ -96,7 +96,7 @@ pub mod encoding {
         };
     }
 
-    /// # The protocol used by the agents to send ad hoc messages to lanes on other agents.
+    /// The protocol used by the agents to send ad hoc messages to lanes on other agents.
     ///
     /// [`crate::AdHocCommand`] messages are sent by agents to the runtime to instruct it to send an ad hoc
     /// command to an arbitrary lane endpoint.
@@ -109,10 +109,6 @@ pub mod encoding {
 
     /// # The protocol used by the runtime to communicate with stores.
     ///
-    /// TODO Non-transient lanes also implicitly contain a store. They should
-    /// ultimately use the initialization component of this protocol. Currently,
-    /// they have initialization messages built into the lane protocol.
-    ///
     /// There are two phases to the communication between the runtime and the agent.
     ///
     /// 1. Initialization
@@ -135,6 +131,11 @@ pub mod encoding {
     /// 2. The store or land sends [`crate::StoreResponse`] messages each time its state
     /// changes which are persisted by the runtime.
     pub mod store {
+
+        // TODO Non-transient lanes also implicitly contain a store. They should
+        // ultimately use the initialization component of this protocol. Currently,
+        // they have initialization messages built into the lane protocol.
+
         pub use crate::store::{
             MapStoreInitDecoder, MapStoreResponseEncoder, RawMapStoreInitDecoder,
             RawMapStoreInitEncoder, RawMapStoreResponseDecoder, RawValueStoreInitDecoder,

api/swimos_agent_protocol/src/model.rs

@@ -58,42 +58,52 @@ impl<T> LaneResponse<T> {
 /// to describe alterations to the lane.
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Form)]
 pub enum MapOperation<K, V> {
+    /// Update the value associated with a key in the map (or insert an entry if they key does not exist).
     #[form(tag = "update")]
     Update {
         key: K,
         #[form(body)]
         value: V,
     },
+    /// Remove an entry from the map, by key (does nothing if there is no such entry).
     #[form(tag = "remove")]
     Remove {
         #[form(header)]
         key: K,
     },
+    /// Remove all entries in the map.
     #[form(tag = "clear")]
     Clear,
 }
 
 /// Representation of map lane messages (used to form the body of Recon messages when operating)
 /// on downlinks. This extends [`MapOperation`] with `Take` (retain the first `n` items) and `Drop`
-/// (remove the first `n` items). We never use these internally but must support them for communicating
-/// with other implementations.
+/// (remove the first `n` items).
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Form, Hash)]
 pub enum MapMessage<K, V> {
+    /// Update the value associated with a key in the map (or insert an entry if they key does not exist).
     #[form(tag = "update")]
     Update {
         key: K,
         #[form(body)]
         value: V,
     },
+    /// Remove an entry from the map, by key (does nothing if there is no such entry).
     #[form(tag = "remove")]
     Remove {
         #[form(header)]
         key: K,
     },
+    /// Remove all entries in the map.
     #[form(tag = "clear")]
     Clear,
+    /// Retain only the first `n` entries in the map, the remainder are removed. The ordering
+    /// used to determine 'first' is the Recon order of the keys. If there are fewer than `n`
+    /// entries in the map, this does nothing.
     #[form(tag = "take")]
     Take(#[form(header_body)] u64),
+    /// Remove the first `n` entries in the map. The ordering used to determine 'first' is the
+    /// Recon order of the keys. If there are fewer than `n` entries in the map, it is cleared.
     #[form(tag = "drop")]
     Drop(#[form(header_body)] u64),
 }
@@ -148,7 +158,8 @@ impl<S, T> AdHocCommand<S, T> {
     /// * `command` - The body of the command message.
     /// * `overwrite_permitted` - Controls the behaviour of command handling in the case of back-pressure.
     /// If this is true, the command maybe be overwritten by a subsequent command to the same target (and so
-    /// will never be sent). If false, the command will be queued instead.
+    /// will never be sent). If false, the command will be queued instead. This is a user specifiable parameter
+    /// in the API.
     pub fn new(address: Address<S>, command: T, overwrite_permitted: bool) -> Self {
         AdHocCommand {
             address,

api/swimos_api/src/agent/lane/mod.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(test)]
+mod tests;
+
 use std::{
     borrow::Borrow,
     fmt::{Display, Formatter},
@@ -72,7 +75,10 @@ impl WarpLaneKind {
 
     pub fn uplink_kind(&self) -> UplinkKind {
         match self {
-            WarpLaneKind::Map | WarpLaneKind::DemandMap | WarpLaneKind::JoinMap => UplinkKind::Map,
+            WarpLaneKind::Map
+            | WarpLaneKind::DemandMap
+            | WarpLaneKind::JoinMap
+            | WarpLaneKind::JoinValue => UplinkKind::Map,
             WarpLaneKind::Supply => UplinkKind::Supply,
             WarpLaneKind::Spatial => todo!("Spatial uplinks not supported."),
             _ => UplinkKind::Value,

api/swimos_api/src/agent/lane/tests.rs

@@ -0,0 +1,36 @@
+use crate::agent::UplinkKind;
+
+use super::WarpLaneKind;
+
+// spatial lanes are omitted as they are unimplemented
+const LANE_KINDS: [WarpLaneKind; 8] = [
+    WarpLaneKind::Command,
+    WarpLaneKind::Demand,
+    WarpLaneKind::DemandMap,
+    WarpLaneKind::Map,
+    WarpLaneKind::JoinMap,
+    WarpLaneKind::JoinValue,
+    WarpLaneKind::Supply,
+    WarpLaneKind::Value,
+];
+
+#[test]
+fn uplink_kinds() {
+    for kind in LANE_KINDS {
+        let uplink_kind = kind.uplink_kind();
+        if kind.map_like() {
+            assert_eq!(uplink_kind, UplinkKind::Map);
+        } else if matches!(kind, WarpLaneKind::Supply) {
+            assert_eq!(uplink_kind, UplinkKind::Supply);
+        } else {
+            assert_eq!(uplink_kind, UplinkKind::Value)
+        }
+    }
+}
+
+// this is here for when spatial lanes are implemented as the test will no longer panic
+#[test]
+#[should_panic]
+fn spatial_uplink_kind() {
+    WarpLaneKind::Spatial.uplink_kind();
+}

example_apps/join_value/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "join_value"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+swimos = { path = "../../swimos", features = ["server", "agent"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+example-util = { path = "../example_util" }

example_apps/join_value/src/building.rs

@@ -0,0 +1,74 @@
+use std::collections::HashMap;
+use swimos::agent::agent_lifecycle::utility::HandlerContext;
+use swimos::agent::event_handler::{EventHandler, HandlerActionExt};
+use swimos::agent::lanes::{CommandLane, JoinValueLane};
+use swimos::agent::projections;
+use swimos::agent::{lifecycle, AgentLaneModel};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct BuildingAgent {
+    lights: JoinValueLane<u64, bool>,
+    register_room: CommandLane<u64>,
+}
+
+#[derive(Clone)]
+pub struct BuildingLifecycle;
+
+#[lifecycle(BuildingAgent)]
+impl BuildingLifecycle {
+    #[on_start]
+    pub fn on_start(
+        &self,
+        context: HandlerContext<BuildingAgent>,
+    ) -> impl EventHandler<BuildingAgent> {
+        context
+            .get_agent_uri()
+            .and_then(move |uri| context.effect(move || println!("Starting agent at: {}", uri)))
+    }
+
+    #[on_stop]
+    pub fn on_stop(
+        &self,
+        context: HandlerContext<BuildingAgent>,
+    ) -> impl EventHandler<BuildingAgent> {
+        context
+            .get_agent_uri()
+            .and_then(move |uri| context.effect(move || println!("Stopping agent at: {}", uri)))
+    }
+
+    #[on_command(register_room)]
+    pub fn register_room(
+        &self,
+        context: HandlerContext<BuildingAgent>,
+        room_id: &u64,
+    ) -> impl EventHandler<BuildingAgent> {
+        let room_id = *room_id;
+        context
+            .get_parameter("name")
+            .and_then(move |building_name: Option<String>| {
+                let building_name = building_name.expect("Missing building name URI parameter");
+                context.add_downlink(
+                    BuildingAgent::LIGHTS,
+                    room_id,
+                    None,
+                    format!("/rooms/{building_name}/{room_id}").as_str(),
+                    "lights",
+                )
+            })
+            .followed_by(context.effect(move || println!("Registered room: {room_id}")))
+    }
+
+    #[on_update(lights)]
+    fn lights(
+        &self,
+        context: HandlerContext<BuildingAgent>,
+        _map: &HashMap<u64, bool>,
+        key: u64,
+        prev: Option<bool>,
+        new_value: &bool,
+    ) -> impl EventHandler<BuildingAgent> {
+        let new_value = *new_value;
+        context.effect(move || println!("Light {key} changed from {prev:?} to {new_value}"))
+    }
+}

example_apps/join_value/src/main.rs

@@ -0,0 +1,56 @@
+// 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, time::Duration};
+
+use crate::{
+    building::{BuildingAgent, BuildingLifecycle},
+    room::{RoomAgent, RoomLifecycle},
+};
+use example_util::{example_logging, manage_handle};
+use swimos::{
+    agent::agent_model::AgentModel, route::RoutePattern, server::Server, server::ServerBuilder,
+};
+
+mod building;
+mod room;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    example_logging()?;
+
+    let building_route = RoutePattern::parse_str("/buildings/:name")?;
+    let building_agent =
+        AgentModel::new(BuildingAgent::default, BuildingLifecycle.into_lifecycle());
+
+    let room_route = RoutePattern::parse_str("/rooms/:building/:room")?;
+    let room_agent = AgentModel::new(RoomAgent::default, RoomLifecycle.into_lifecycle());
+
+    let server = ServerBuilder::with_plane_name("Building Plane")
+        .add_route(building_route, building_agent)
+        .add_route(room_route, room_agent)
+        .update_config(|config| {
+            config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
+        })
+        .build()
+        .await?;
+
+    let (task, handle) = server.run();
+    let shutdown = manage_handle(handle);
+    let (_, result) = tokio::join!(shutdown, task);
+
+    result?;
+    println!("Server stopped successfully.");
+    Ok(())
+}

example_apps/join_value/src/room.rs

@@ -0,0 +1,52 @@
+use std::str::FromStr;
+use swimos::agent::agent_lifecycle::utility::HandlerContext;
+use swimos::agent::event_handler::{EventHandler, HandlerActionExt};
+use swimos::agent::lanes::ValueLane;
+use swimos::agent::projections;
+use swimos::agent::{lifecycle, AgentLaneModel};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct RoomAgent {
+    lights: ValueLane<bool>,
+}
+
+#[derive(Clone)]
+pub struct RoomLifecycle;
+
+#[lifecycle(RoomAgent)]
+impl RoomLifecycle {
+    #[on_start]
+    pub fn on_start(&self, context: HandlerContext<RoomAgent>) -> impl EventHandler<RoomAgent> {
+        context
+            .get_agent_uri()
+            .and_then(move |uri| context.effect(move || println!("Starting agent at: {}", uri)))
+            .followed_by(context.get_parameter("building").and_then(
+                move |building_name: Option<String>| {
+                    context
+                        .get_parameter("room")
+                        .and_then(move |room_id: Option<String>| {
+                            let building_name =
+                                building_name.expect("Missing building URI parameter");
+                            let room_id_str = room_id.expect("Missing room ID URI parameter");
+                            let room_id = u64::from_str(room_id_str.as_str())
+                                .expect("Expected a u64 room ID");
+
+                            context.send_command(
+                                None,
+                                format!("/buildings/{building_name}"),
+                                "register_room".to_string(),
+                                room_id,
+                            )
+                        })
+                },
+            ))
+    }
+
+    #[on_stop]
+    pub fn on_stop(&self, context: HandlerContext<RoomAgent>) -> impl EventHandler<RoomAgent> {
+        context
+            .get_agent_uri()
+            .and_then(move |uri| context.effect(move || println!("Starting agent at: {}", uri)))
+    }
+}