Merge pull request #581 from swimos/supply-lane

Supply lanes

Author: horned-sphere

Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "example_apps/command_lane",
     "example_apps/value_store",
     "example_apps/map_store",
+    "example_apps/supply_lane",
     "example_apps/value_lane_persistence",
     "example_apps/map_lane_persistence",
     "example_apps/value_store_persistence",

docs/define.md

@@ -29,6 +29,7 @@ The derive macro for `AgentLaneModel` can be applied to any struct type where al
 * Join Value Lanes: `swim::agent::lanes::JoinValueLane`.
 * Join Map Lanes: `swim::agent::lanes::JoinMapLane`.
 * HTTP Lanes: `swim::agent::lanes::HttpLane` (or the shorthand `swim::agent::lanes::SimpleHttpLane`).
+* Supply Lanes: `swim::agent::lanes::SupplyLane`.
 
 The supported store types are:
 

docs/lifecycle.md

@@ -137,6 +137,11 @@ fn my_demand_keys_handler(&self, context: HandlerContext<ExampleAgent>) -> impl
 }
 ```
 
+Supply lane events
+-------------------
+
+Supply lanes generate no events on a lifecycle.
+
 Value lane events
 -----------------
 

example_apps/supply_lane/Cargo.toml

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

example_apps/supply_lane/src/agent.rs

@@ -0,0 +1,68 @@
+// 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 swim::agent::lanes::SupplyLane;
+use swim::agent::{
+    agent_lifecycle::utility::HandlerContext,
+    event_handler::{EventHandler, HandlerActionExt},
+    lanes::ValueLane,
+    lifecycle, projections, AgentLaneModel,
+};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct ExampleAgent {
+    lane: ValueLane<i32>,
+    supply: SupplyLane<i32>,
+}
+
+#[derive(Clone)]
+pub struct ExampleLifecycle;
+
+#[lifecycle(ExampleAgent)]
+impl ExampleLifecycle {
+    #[on_start]
+    pub fn on_start(
+        &self,
+        context: HandlerContext<ExampleAgent>,
+    ) -> impl EventHandler<ExampleAgent> {
+        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<ExampleAgent>,
+    ) -> impl EventHandler<ExampleAgent> {
+        context.get_agent_uri().and_then(move |uri| {
+            context.effect(move || {
+                println!("Stopping agent at: {}", uri);
+            })
+        })
+    }
+
+    #[on_event(lane)]
+    pub fn on_event(
+        &self,
+        context: HandlerContext<ExampleAgent>,
+        value: &i32,
+    ) -> impl EventHandler<ExampleAgent> {
+        let n = *value;
+        context.supply(ExampleAgent::SUPPLY, n * 2)
+    }
+}

example_apps/supply_lane/src/main.rs

@@ -0,0 +1,53 @@
+// 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 example_util::{example_logging, manage_handle};
+use swim::{
+    agent::agent_model::AgentModel,
+    route::RoutePattern,
+    server::{Server, ServerBuilder},
+};
+
+use crate::agent::{ExampleAgent, ExampleLifecycle};
+
+mod agent;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    example_logging()?;
+    let route = RoutePattern::parse_str("/example/:name}")?;
+
+    let lifecycle = ExampleLifecycle;
+    let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
+
+    let server = ServerBuilder::with_plane_name("Example Plane")
+        .add_route(route, 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(())
+}

server/swim_agent/src/agent_lifecycle/utility/mod.rs

@@ -50,6 +50,7 @@ use crate::lanes::demand::{Cue, DemandLane};
 use crate::lanes::demand_map::CueKey;
 use crate::lanes::join_map::JoinMapAddDownlink;
 use crate::lanes::join_value::{JoinValueAddDownlink, JoinValueLane};
+use crate::lanes::supply::{Supply, SupplyLane};
 use crate::lanes::{DemandMapLane, JoinMapLane};
 
 pub use self::downlink_builder::event::{
@@ -366,6 +367,22 @@ impl<Agent: 'static> HandlerContext<Agent> {
         CueKey::new(lane, key)
     }
 
+    /// Create an event handler that will supply an event to a supply lane.
+    ///
+    /// #Arguments
+    /// * `lane` - Projection to the supply lane.
+    /// * `value` - The value to supply.
+    pub fn supply<V>(
+        &self,
+        lane: fn(&Agent) -> &SupplyLane<V>,
+        value: V,
+    ) -> impl EventHandler<Agent> + Send + 'static
+    where
+        V: Send + 'static,
+    {
+        Supply::new(lane, value)
+    }
+
     /// Suspend a future to be executed by the agent task. The future must result in another
     /// event handler that will be executed by the agent upon completion.
     pub fn suspend<Fut, H>(&self, future: Fut) -> impl EventHandler<Agent> + Send + 'static

server/swim_agent/src/lanes/mod.rs

@@ -19,6 +19,8 @@ pub mod http;
 mod join;
 pub mod map;
 mod queues;
+
+pub mod supply;
 pub mod value;
 
 pub use join::map as join_map;
@@ -38,6 +40,7 @@ pub use self::{
     join_map::JoinMapLane,
     join_value::JoinValueLane,
     map::MapLane,
+    supply::SupplyLane,
     value::ValueLane,
 };