Repository: swimos/swim-rust (language: rust)

Issue:
Add support for supply lanes.
#506 is a prerequisite for this issue.

Add supply lanes to `swim_agent::lanes` (see existing value and map lanes in this module for comparison).

Supply lanes should behave in the same way to their Java equivalent.

A demand lane has an associated type `T` and has no associated event handlers.

There should be an associated `EventHandler` which pushes a value of type `T` into the lane. When this is executed, it will cause the value to be scheduled to be written to the output buffer. Note that supply lanes should not have backpressure relief so multiple values generated in a single event handler execution need to be queued rather than overwriting. Care should be taken to manage the capacity of the queue.


Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 77f8917cb..8a6f54bec 100644
--- a/Cargo.toml
+++ b/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",
diff --git a/docs/define.md b/docs/define.md
index c2361c55f..87d8eb36f 100644
--- a/docs/define.md
+++ b/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:
 
diff --git a/docs/lifecycle.md b/docs/lifecycle.md
index 4bccab554..b92e33078 100644
--- a/docs/lifecycle.md
+++ b/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
 -----------------
 
diff --git a/example_apps/supply_lane/Cargo.toml b/example_apps/supply_lane/Cargo.toml
new file mode 100644
index 000000000..26312d341
--- /dev/null
+++ b/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" }
diff --git a/example_apps/supply_lane/src/agent.rs b/example_apps/supply_lane/src/agent.rs
new file mode 100644
index 000000000..8a8553b96
--- /dev/null
+++ b/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)
+    }
+}
diff --git a/example_apps/supply_lane/src/main.rs b/example_apps/supply_lane/src/main.rs
new file mode 100644
index 000000000..72fda5c69
--- /dev/null
+++ b/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(())
+}
diff --git a/server/swim_agent/src/agent_lifecycle/utility/mod.rs b/server/swim_agent/src/agent_lifecycle/utility/mod.rs
index 367b18ece..103af8b51 100644
--- a/server/swim_agent/src/agent_lifecycle/utility/mod.rs
+++ b/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
diff --git a/server/swim_agent/src/lanes/mod.rs b/server/swim_agent/src/lanes/mod.rs
index eb085e794..e018fc188 100644
--- a/server/swim_agent/src/lanes/mod.rs
+++ b/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,
 };
 
diff --git a/server/swim_agent/src/lanes/supply/mod.rs b/server/swim_agent/src/lanes/supply/mod.rs
new file mode 100644
index 000000000..b1279f7fa
--- /dev/null
+++ b/server/swim_agent/src/lanes/supply/mod.rs
@@ -0,0 +1,202 @@
+// 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::{cell::RefCell, collections::VecDeque};
+
+use bytes::BytesMut;
+use static_assertions::assert_impl_all;
+use tokio_util::codec::Encoder;
+use uuid::Uuid;
+
+use swim_api::protocol::agent::{LaneResponse, ValueLaneResponseEncoder};
+use swim_form::structural::write::StructuralWritable;
+
+use crate::event_handler::EventHandlerError;
+use crate::{
+    agent_model::WriteResult,
+    event_handler::{ActionContext, HandlerAction, Modification, StepResult},
+    item::AgentItem,
+    meta::AgentMetadata,
+};
+
+use super::LaneItem;
+
+#[cfg(test)]
+mod tests;
+
+#[derive(Debug)]
+struct SupplyLaneInner<T> {
+    sync_queue: VecDeque<Uuid>,
+    event_queue: VecDeque<T>,
+}
+
+impl<T> Default for SupplyLaneInner<T> {
+    fn default() -> SupplyLaneInner<T> {
+        SupplyLaneInner {
+            sync_queue: Default::default(),
+            event_queue: Default::default(),
+        }
+    }
+}
+
+/// A stateless lane that pushes events received directly to all uplinks attached to it.
+///
+/// A Supply lane can push a value by executing an instance of [`Supply`] (which can be constructed
+/// using the [`crate::agent_lifecycle::utility::HandlerContext`]).
+#[derive(Debug)]
+pub struct SupplyLane<T> {
+    id: u64,
+    inner: RefCell<SupplyLaneInner<T>>,
+}
+
+impl<T> SupplyLane<T> {
+    /// Create a Supply lane with the specified ID (this needs to be unique within an agent).
+    pub fn new(id: u64) -> Self {
+        SupplyLane {
+            id,
+            inner: Default::default(),
+        }
+    }
+
+    fn push(&self, event: T) {
+        let mut guard = self.inner.borrow_mut();
+        guard.event_queue.push_back(event)
+    }
+
+    fn sync(&self, id: Uuid) {
+        let mut guard = self.inner.borrow_mut();
+        guard.sync_queue.push_back(id);
+    }
+}
+
+assert_impl_all!(SupplyLane<()>: Send);
+
+impl<T> AgentItem for SupplyLane<T> {
+    fn id(&self) -> u64 {
+        self.id
+    }
+}
+
+const INFALLIBLE_SER: &str = "Serializing a value to recon should be infallible.";
+
+impl<T> LaneItem for SupplyLane<T>
+where
+    T: StructuralWritable,
+{
+    fn write_to_buffer(&self, buffer: &mut BytesMut) -> WriteResult {
+        let SupplyLane { inner, .. } = self;
+        let mut guard = inner.borrow_mut();
+        let SupplyLaneInner {
+            sync_queue,
+            event_queue,
+        } = &mut *guard;
+
+        match sync_queue.pop_front() {
+            Some(remote) => {
+                let mut encoder = ValueLaneResponseEncoder::default();
+                encoder
+                    .encode(LaneResponse::<T>::synced(remote), buffer)
+                    .expect(INFALLIBLE_SER);
+            }
+            None => {
+                if let Some(event) = event_queue.pop_front() {
+                    let mut encoder = ValueLaneResponseEncoder::default();
+                    encoder
+                        .encode(LaneResponse::event(event), buffer)
+                        .expect(INFALLIBLE_SER);
+                }
+            }
+        }
+
+        if !event_queue.is_empty() || !sync_queue.is_empty() {
+            WriteResult::DataStillAvailable
+        } else {
+            WriteResult::Done
+        }
+    }
+}
+
+pub struct Supply<Context, T> {
+    projection: fn(&Context) -> &SupplyLane<T>,
+    value: Option<T>,
+}
+
+impl<Context, T> Supply<Context, T> {
+    pub fn new(projection: fn(&Context) -> &SupplyLane<T>, event: T) -> Self {
+        Supply {
+            projection,
+            value: Some(event),
+        }
+    }
+}
+
+impl<Context, T> HandlerAction<Context> for Supply<Context, T> {
+    type Completion = ();
+
+    fn step(
+        &mut self,
+        _action_context: &mut ActionContext<Context>,
+        _meta: AgentMetadata,
+        context: &Context,
+    ) -> StepResult<Self::Completion> {
+        let Supply { projection, value } = self;
+        if let Some(value) = value.take() {
+            let lane = projection(context);
+            lane.push(value);
+            StepResult::Complete {
+                modified_item: Some(Modification::no_trigger(lane.id())),
+                result: (),
+            }
+        } else {
+            StepResult::Fail(EventHandlerError::SteppedAfterComplete)
+        }
+    }
+}
+
+pub struct SupplyLaneSync<Context, T> {
+    projection: fn(&Context) -> &SupplyLane<T>,
+    id: Option<Uuid>,
+}
+
+impl<Context, T> SupplyLaneSync<Context, T> {
+    pub fn new(projection: fn(&Context) -> &SupplyLane<T>, id: Uuid) -> Self {
+        SupplyLaneSync {
+            projection,
+            id: Some(id),
+        }
+    }
+}
+
+impl<Context, T> HandlerAction<Context> for SupplyLaneSync<Context, T> {
+    type Completion = ();
+
+    fn step(
+        &mut self,
+        _action_context: &mut ActionContext<Context>,
+        _meta: AgentMetadata,
+        context: &Context,
+    ) -> StepResult<Self::Completion> {
+        let SupplyLaneSync { projection, id } = self;
+        if let Some(id) = id.take() {
+            let lane = projection(context);
+            lane.sync(id);
+            StepResult::Complete {
+                modified_item: Some(Modification::no_trigger(lane.id)),
+                result: (),
+            }
+        } else {
+            StepResult::after_done()
+        }
+    }
+}
diff --git a/server/swim_agent_derive/src/lane_model_derive/mod.rs b/server/swim_agent_derive/src/lane_model_derive/mod.rs
index 523ae910c..48358cca2 100644
--- a/server/swim_agent_derive/src/lane_model_derive/mod.rs
+++ b/server/swim_agent_derive/src/lane_model_derive/mod.rs
@@ -407,6 +407,9 @@ impl<'a> FieldInitializer<'a> {
             ItemSpec::DemandMap(_, _) => {
                 quote!(#name: #root::lanes::DemandMapLane::new(#ordinal))
             }
+            ItemSpec::Supply(_) => {
+                quote!(#name: #root::lanes::SupplyLane::new(#ordinal))
+            }
             ItemSpec::Value(ItemKind::Lane, _) => {
                 quote!(#name: #root::lanes::ValueLane::new(#ordinal, ::core::default::Default::default()))
             }
@@ -459,7 +462,8 @@ impl<'a> HandlerType<'a> {
             WarpLaneSpec::Demand(_)
             | WarpLaneSpec::DemandMap(_, _)
             | WarpLaneSpec::JoinValue(_, _)
-            | WarpLaneSpec::JoinMap(_, _, _) => {
+            | WarpLaneSpec::JoinMap(_, _, _)
+            | WarpLaneSpec::Supply(_) => {
                 quote!(#root::event_handler::UnitHandler)
             }
         }
@@ -494,6 +498,9 @@ impl<'a> SyncHandlerType<'a> {
             WarpLaneSpec::JoinMap(l, k, v) => {
                 quote!(#root::lanes::join_map::JoinMapLaneSync<#agent_name, #l, #k, #v>)
             }
+            WarpLaneSpec::Supply(t) => {
+                quote!(#root::lanes::supply::SupplyLaneSync<#agent_name, #t>)
+            }
         }
     }
 }
@@ -553,7 +560,8 @@ impl<'a> WarpLaneHandlerMatch<'a> {
             WarpLaneSpec::Demand(_)
             | WarpLaneSpec::DemandMap(_, _)
             | WarpLaneSpec::JoinValue(_, _)
-            | WarpLaneSpec::JoinMap(_, _, _) => {
+            | WarpLaneSpec::JoinMap(_, _, _)
+            | WarpLaneSpec::Supply(_) => {
                 quote!(#root::event_handler::UnitHandler::default())
             }
         };
@@ -659,6 +667,9 @@ impl<'a> SyncHandlerMatch<'a> {
             WarpLaneSpec::JoinMap(l, k, v) => {
                 quote!(#root::lanes::join_map::JoinMapLaneSync::<#agent_name, #l, #k, #v>::new(|agent: &#agent_name| &agent.#name, id))
             }
+            WarpLaneSpec::Supply(ty) => {
+                quote!(#root::lanes::supply::SupplyLaneSync::<#agent_name, #ty>::new(|agent: &#agent_name| &agent.#name, id))
+            }
         };
         quote! {
             #name_lit => {
@@ -810,6 +821,9 @@ impl<'a> LaneSpecInsert<'a> {
                 quote!(#root::agent_model::ItemDescriptor::WarpLane { kind: #root::agent_model::WarpLaneKind::JoinMap, flags: #flags })
             }
             ItemSpec::Http(_) => quote!(#root::agent_model::ItemDescriptor::Http),
+            ItemSpec::Supply(_) => {
+                quote!(#root::agent_model::ItemDescriptor::WarpLane { kind: #root::agent_model::WarpLaneKind::Supply, flags: #flags })
+            }
         };
         let external_lane_name = model.external_literal();
         let lifecycle_lane_name = model.lifecycle_literal();
diff --git a/server/swim_agent_derive/src/lane_model_derive/model.rs b/server/swim_agent_derive/src/lane_model_derive/model.rs
index dec543f82..5898d56d8 100644
--- a/server/swim_agent_derive/src/lane_model_derive/model.rs
+++ b/server/swim_agent_derive/src/lane_model_derive/model.rs
@@ -103,6 +103,7 @@ pub enum ItemSpec<'a> {
     DemandMap(&'a Type, &'a Type),
     Value(ItemKind, &'a Type),
     Map(ItemKind, &'a Type, &'a Type),
+    Supply(&'a Type),
     JoinValue(&'a Type, &'a Type),
     JoinMap(&'a Type, &'a Type, &'a Type),
     Http(HttpLaneSpec<'a>),
@@ -118,6 +119,7 @@ impl<'a> ItemSpec<'a> {
             ItemSpec::Map(ItemKind::Lane, k, v) => Some(WarpLaneSpec::Map(k, v)),
             ItemSpec::JoinValue(k, v) => Some(WarpLaneSpec::JoinValue(k, v)),
             ItemSpec::JoinMap(l, k, v) => Some(WarpLaneSpec::JoinMap(l, k, v)),
+            ItemSpec::Supply(t) => Some(WarpLaneSpec::Supply(t)),
             _ => None,
         }
     }
@@ -140,6 +142,7 @@ impl<'a> ItemSpec<'a> {
             ItemSpec::Demand(_) => ItemKind::Lane,
             ItemSpec::DemandMap(_, _) => ItemKind::Lane,
             ItemSpec::Http(_) => ItemKind::Lane,
+            ItemSpec::Supply(_) => ItemKind::Lane,
         }
     }
 }
@@ -151,6 +154,7 @@ pub enum WarpLaneSpec<'a> {
     Demand(&'a Type),
     DemandMap(&'a Type, &'a Type),
     Value(&'a Type),
+    Supply(&'a Type),
     Map(&'a Type, &'a Type),
     JoinValue(&'a Type, &'a Type),
     JoinMap(&'a Type, &'a Type, &'a Type),
@@ -166,7 +170,10 @@ pub struct HttpLaneSpec<'a> {
 
 impl<'a> ItemSpec<'a> {
     pub fn is_stateful(&self) -> bool {
-        !matches!(self, ItemSpec::Command(_) | ItemSpec::Demand(_))
+        !matches!(
+            self,
+            ItemSpec::Command(_) | ItemSpec::Demand(_) | ItemSpec::Supply(_)
+        )
     }
 }
 
@@ -345,6 +352,7 @@ const MAP_LANE_NAME: &str = "MapLane";
 const MAP_STORE_NAME: &str = "MapStore";
 const JOIN_VALUE_LANE_NAME: &str = "JoinValueLane";
 const JOIN_MAP_LANE_NAME: &str = "JoinMapLane";
+const SUPPLY_LANE_NAME: &str = "SupplyLane";
 const HTTP_LANE_NAME: &str = "HttpLane";
 const SIMPLE_HTTP_LANE_NAME: &str = "SimpleHttpLane";
 
@@ -453,6 +461,17 @@ fn extract_lane_model(field: &Field) -> Validation<ItemModel<'_>, Errors<syn::Er
                             )),
                             Err(e) => Validation::fail(Errors::of(e)),
                         },
+                        SUPPLY_LANE_NAME => {
+                            match single_param(arguments) {
+                                Ok(param) => Validation::valid(ItemModel::new(
+                                    fld_name,
+                                    ItemSpec::Supply(param),
+                                    ItemFlags::TRANSIENT, //Supply lanes are always transient.
+                                    transform,
+                                )),
+                                Err(e) => Validation::fail(Errors::of(e)),
+                            }
+                        }
                         name @ (HTTP_LANE_NAME | SIMPLE_HTTP_LANE_NAME) => {
                             match http_params(arguments, name == SIMPLE_HTTP_LANE_NAME) {
                                 Ok(spec) => Validation::valid(ItemModel::new(
diff --git a/swim/src/agent.rs b/swim/src/agent.rs
index a5d60c328..852438dfe 100644
--- a/swim/src/agent.rs
+++ b/swim/src/agent.rs
@@ -244,7 +244,7 @@ pub mod item {
 pub mod lanes {
     pub use swim_agent::lanes::{
         CommandLane, DemandLane, DemandMapLane, HttpLane, JoinMapLane, JoinValueLane, LaneItem,
-        LinkClosedResponse, MapLane, SimpleHttpLane, ValueLane,
+        LinkClosedResponse, MapLane, SimpleHttpLane, SupplyLane, ValueLane,
     };
 
     pub mod command {
@@ -311,6 +311,10 @@ pub mod lanes {
             pub use swim_agent::lanes::http::lifecycle::StatefulHttpLaneLifecycle;
         }
     }
+
+    pub mod supply {
+        pub use swim_agent::lanes::supply::SupplyLaneSync;
+    }
 }
 
 pub mod state {


Test patch:
diff --git a/server/swim_agent/src/lanes/supply/tests.rs b/server/swim_agent/src/lanes/supply/tests.rs
new file mode 100644
index 000000000..ad7ef96f9
--- /dev/null
+++ b/server/swim_agent/src/lanes/supply/tests.rs
@@ -0,0 +1,244 @@
+// 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 bytes::BytesMut;
+use tokio_util::codec::Decoder;
+use uuid::Uuid;
+
+use swim_api::{
+    agent::AgentConfig,
+    protocol::agent::{LaneResponse, ValueLaneResponseDecoder},
+};
+use swim_utilities::routing::route_uri::RouteUri;
+
+use crate::lanes::supply::{Supply, SupplyLaneSync};
+use crate::{
+    agent_model::WriteResult,
+    event_handler::{EventHandlerError, HandlerAction, Modification, StepResult},
+    lanes::LaneItem,
+    meta::AgentMetadata,
+    test_context::dummy_context,
+};
+
+use super::SupplyLane;
+
+const CONFIG: AgentConfig = AgentConfig::DEFAULT;
+const NODE_URI: &str = "/node";
+const LANE_ID: u64 = 38;
+const SYNC_ID: Uuid = Uuid::from_u128(77);
+const SYNC_ID2: Uuid = Uuid::from_u128(929);
+
+fn make_uri() -> RouteUri {
+    RouteUri::try_from(NODE_URI).expect("Bad URI.")
+}
+
+fn make_meta<'a>(
+    uri: &'a RouteUri,
+    route_params: &'a HashMap<String, String>,
+) -> AgentMetadata<'a> {
+    AgentMetadata::new(uri, route_params, &CONFIG)
+}
+
+struct TestAgent {
+    lane: SupplyLane<i32>,
+}
+
+impl Default for TestAgent {
+    fn default() -> Self {
+        Self {
+            lane: SupplyLane::new(LANE_ID),
+        }
+    }
+}
+
+impl TestAgent {
+    const LANE: fn(&TestAgent) -> &SupplyLane<i32> = |agent| &agent.lane;
+}
+
+#[test]
+fn supply_event_handler() {
+    let uri = make_uri();
+    let route_params = HashMap::new();
+    let meta = make_meta(&uri, &route_params);
+    let agent = TestAgent::default();
+
+    let mut handler = Supply::new(TestAgent::LANE, 13);
+
+    let result = handler.step(
+        &mut dummy_context(&mut HashMap::new(), &mut BytesMut::new()),
+        meta,
+        &agent,
+    );
+
+    let _modification = Modification::no_trigger(LANE_ID);
+    assert!(matches!(
+        result,
+        StepResult::Complete {
+            modified_item: Some(_modification),
+            result: ()
+        }
+    ));
+
+    let result = handler.step(
+        &mut dummy_context(&mut HashMap::new(), &mut BytesMut::new()),
+        meta,
+        &agent,
+    );
+    assert!(matches!(
+        result,
+        StepResult::Fail(EventHandlerError::SteppedAfterComplete)
+    ));
+}
+
+#[test]
+fn supply_lane_sync_handler() {
+    let uri = make_uri();
+    let route_params = HashMap::new();
+    let meta = make_meta(&uri, &route_params);
+    let agent = TestAgent::default();
+
+    let mut handler = SupplyLaneSync::new(TestAgent::LANE, SYNC_ID);
+
+    let result = handler.step(
+        &mut dummy_context(&mut HashMap::new(), &mut BytesMut::new()),
+        meta,
+        &agent,
+    );
+
+    let _modification = Modification::no_trigger(LANE_ID);
+    assert!(matches!(
+        result,
+        StepResult::Complete {
+            modified_item: Some(_modification),
+            result: ()
+        }
+    ));
+
+    let guard = agent.lane.inner.borrow();
+    assert_eq!(
+        guard.sync_queue.iter().copied().collect::<Vec<_>>(),
+        vec![SYNC_ID]
+    );
+
+    let result = handler.step(
+        &mut dummy_context(&mut HashMap::new(), &mut BytesMut::new()),
+        meta,
+        &agent,
+    );
+    assert!(matches!(
+        result,
+        StepResult::Fail(EventHandlerError::SteppedAfterComplete)
+    ));
+}
+
+fn read_buffer(buffer: &mut BytesMut) -> Vec<LaneResponse<i32>> {
+    let mut decoder = ValueLaneResponseDecoder::default();
+    let mut results = vec![];
+    while !buffer.is_empty() {
+        let mut r = decoder.decode(buffer).expect("Decode failed.");
+        if r.is_none() {
+            r = decoder.decode_eof(buffer).expect("Decode failed.");
+        }
+        let msg = r.expect("Incomplete record");
+        results.push(match msg {
+            LaneResponse::StandardEvent(body) => LaneResponse::StandardEvent(
+                std::str::from_utf8(body.as_ref())
+                    .expect("Bad utf.")
+                    .parse::<i32>()
+                    .expect("Bad body."),
+            ),
+            LaneResponse::Initialized => LaneResponse::Initialized,
+            LaneResponse::SyncEvent(id, body) => LaneResponse::SyncEvent(
+                id,
+                std::str::from_utf8(body.as_ref())
+                    .expect("Bad utf.")
+                    .parse::<i32>()
+                    .expect("Bad body."),
+            ),
+            LaneResponse::Synced(id) => LaneResponse::Synced(id),
+        });
+    }
+    results
+}
+
+#[test]
+fn write_supplied() {
+    let lane = SupplyLane::new(LANE_ID);
+    lane.push(13);
+
+    let mut buffer = BytesMut::new();
+    assert_eq!(lane.write_to_buffer(&mut buffer), WriteResult::Done);
+
+    let messages = read_buffer(&mut buffer);
+
+    assert_eq!(messages, vec![LaneResponse::StandardEvent(13)]);
+    assert!(read_buffer(&mut buffer).is_empty());
+}
+
+#[test]
+fn write_sync() {
+    let lane = SupplyLane::<i32>::new(LANE_ID);
+    lane.sync(SYNC_ID);
+
+    let mut buffer = BytesMut::new();
+    assert_eq!(lane.write_to_buffer(&mut buffer), WriteResult::Done);
+
+    let messages = read_buffer(&mut buffer);
+
+    assert_eq!(messages, vec![LaneResponse::Synced(SYNC_ID),]);
+    assert!(read_buffer(&mut buffer).is_empty());
+}
+
+#[test]
+fn write_multiple() {
+    let lane = SupplyLane::new(LANE_ID);
+    lane.push(13);
+    lane.sync(SYNC_ID);
+    lane.push(14);
+    lane.sync(SYNC_ID2);
+    lane.push(15);
+
+    let mut buffer = BytesMut::new();
+    let mut result = WriteResult::DataStillAvailable;
+
+    while result == WriteResult::DataStillAvailable {
+        result = lane.write_to_buffer(&mut buffer);
+    }
+
+    let messages = read_buffer(&mut buffer);
+
+    assert_eq!(
+        messages,
+        vec![
+            LaneResponse::Synced(SYNC_ID),
+            LaneResponse::Synced(SYNC_ID2),
+            LaneResponse::StandardEvent(13),
+            LaneResponse::StandardEvent(14),
+            LaneResponse::StandardEvent(15),
+        ]
+    );
+
+    assert!(read_buffer(&mut buffer).is_empty());
+}
+
+#[test]
+fn read_empty() {
+    let lane = SupplyLane::<i32>::new(LANE_ID);
+    let mut buffer = BytesMut::new();
+
+    lane.write_to_buffer(&mut buffer);
+    assert!(read_buffer(&mut buffer).is_empty());
+}
diff --git a/swim/tests/deriveagentlanemodel.rs b/swim/tests/deriveagentlanemodel.rs
index 709ab8741..945257de3 100644
--- a/swim/tests/deriveagentlanemodel.rs
+++ b/swim/tests/deriveagentlanemodel.rs
@@ -13,8 +13,8 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-
 use std::fmt::Write;
+
 use swim::agent::agent_model::ItemFlags;
 use swim::agent::lanes::{CommandLane, MapLane, ValueLane};
 use swim::agent::model::MapMessage;
@@ -25,7 +25,7 @@ use swim::agent::AgentLaneModel;
 use swim_agent::agent_model::{ItemDescriptor, ItemSpec};
 use swim_agent::lanes::http::Recon;
 use swim_agent::lanes::{
-    DemandLane, DemandMapLane, HttpLane, JoinMapLane, JoinValueLane, SimpleHttpLane,
+    DemandLane, DemandMapLane, HttpLane, JoinMapLane, JoinValueLane, SimpleHttpLane, SupplyLane,
 };
 use swim_agent::reexport::bytes::Bytes;
 use swim_agent::stores::{MapStore, ValueStore};
@@ -229,6 +229,16 @@ fn single_demand_map_lane() {
     check_agent::<SingleDemandMapLane>(vec![transient_lane(0, "lane", WarpLaneKind::DemandMap)]);
 }
 
+#[test]
+fn single_supply_lane() {
+    #[derive(AgentLaneModel)]
+    struct SingleSupplyLane {
+        lane: SupplyLane<i32>,
+    }
+
+    check_agent::<SingleSupplyLane>(vec![transient_lane(0, "lane", WarpLaneKind::Supply)]);
+}
+
 #[test]
 fn two_value_lanes() {
     #[derive(AgentLaneModel)]
@@ -313,6 +323,20 @@ fn two_demand_lanes() {
     ]);
 }
 
+#[test]
+fn two_supply_lanes() {
+    #[derive(AgentLaneModel)]
+    struct TwoSupplyLanes {
+        first: SupplyLane<i32>,
+        second: SupplyLane<i32>,
+    }
+
+    check_agent::<TwoSupplyLanes>(vec![
+        transient_lane(0, "first", WarpLaneKind::Supply),
+        transient_lane(1, "second", WarpLaneKind::Supply),
+    ]);
+}
+
 #[test]
 fn two_demand_map_lanes() {
     #[derive(AgentLaneModel)]
@@ -369,19 +393,21 @@ fn multiple_lanes() {
         eighth: DemandMapLane<i32, i32>,
         ninth: SimpleHttpLane<i32>,
         tenth: JoinMapLane<i32, i32, i32>,
+        eleventh: SupplyLane<i32>,
     }
 
     check_agent::<MultipleLanes>(vec![
         persistent_lane(0, "first", WarpLaneKind::Value),
-        persistent_lane(2, "third", WarpLaneKind::Value),
-        transient_lane(4, "fifth", WarpLaneKind::Command),
-        transient_lane(6, "seventh", WarpLaneKind::Demand),
         persistent_lane(1, "second", WarpLaneKind::Map),
+        persistent_lane(2, "third", WarpLaneKind::Value),
         persistent_lane(3, "fourth", WarpLaneKind::Map),
+        transient_lane(4, "fifth", WarpLaneKind::Command),
         transient_lane(5, "sixth", WarpLaneKind::JoinValue),
+        transient_lane(6, "seventh", WarpLaneKind::Demand),
         transient_lane(7, "eighth", WarpLaneKind::DemandMap),
         http_lane(8, "ninth"),
         transient_lane(9, "tenth", WarpLaneKind::JoinMap),
+        transient_lane(10, "eleventh", WarpLaneKind::Supply),
     ]);
 }
 
@@ -493,6 +519,21 @@ fn demand_lane_tagged_transient() {
     ]);
 }
 
+#[test]
+fn supply_lane_tagged_transient() {
+    #[derive(AgentLaneModel)]
+    struct TwoSupplyLanes {
+        #[lane(transient)]
+        first: SupplyLane<i32>,
+        second: SupplyLane<i32>,
+    }
+
+    check_agent::<TwoSupplyLanes>(vec![
+        transient_lane(0, "first", WarpLaneKind::Supply),
+        transient_lane(1, "second", WarpLaneKind::Supply),
+    ]);
+}
+
 #[test]
 fn demand_map_lane_tagged_transient() {
     #[derive(AgentLaneModel)]
@@ -659,11 +700,12 @@ fn two_general_http_lanes() {
 }
 
 mod isolated {
+    use swim_api::lane::WarpLaneKind;
+    use swim_api::store::StoreKind;
+
     use crate::check_agent;
 
     use super::{http_lane, persistent_lane, persistent_store, transient_lane};
-    use swim_api::lane::WarpLaneKind;
-    use swim_api::store::StoreKind;
 
     #[test]
     fn multiple_items_qualified() {