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

Issue:
Example applications missing client binaries
The example applications should have client cargo examples


Discussion / hints:

Fix patch:
diff --git a/example_apps/aggregations/Cargo.toml b/example_apps/aggregations/Cargo.toml
index 64d45ae00..2d93727ea 100644
--- a/example_apps/aggregations/Cargo.toml
+++ b/example_apps/aggregations/Cargo.toml
@@ -6,8 +6,12 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 swimos_form = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 rand = { workspace = true }
-futures-util = { workspace = true }
\ No newline at end of file
+futures-util = { workspace = true }
+
+[[bin]]
+name = "aggregations_client"
\ No newline at end of file
diff --git a/example_apps/aggregations/src/bin/aggregations_client.rs b/example_apps/aggregations/src/bin/aggregations_client.rs
new file mode 100644
index 000000000..b569e5b8f
--- /dev/null
+++ b/example_apps/aggregations/src/bin/aggregations_client.rs
@@ -0,0 +1,40 @@
+use std::error::Error;
+use swimos_client::{BasicValueDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let car_lifecycle = BasicValueDownlinkLifecycle::default()
+        .on_event_blocking(|speed| println!("/cars/1 speed: {speed}"));
+    client_handle
+        .value_downlink::<u64>(RemotePath::new(host, "/cars/1", "speed"))
+        .lifecycle(car_lifecycle)
+        .open()
+        .await?;
+
+    let area_lifecycle = BasicValueDownlinkLifecycle::default()
+        .on_event_blocking(|speed| println!("/area/arbury average speed: {speed}"));
+    client_handle
+        .value_downlink::<f64>(RemotePath::new(host, "/area/arbury", "average_speed"))
+        .lifecycle(area_lifecycle)
+        .open()
+        .await?;
+
+    let city_lifecycle = BasicValueDownlinkLifecycle::default()
+        .on_event_blocking(|speed| println!("/city average speed: {speed}"));
+    client_handle
+        .value_downlink::<f64>(RemotePath::new(host, "/city", "average_speed"))
+        .lifecycle(city_lifecycle)
+        .open()
+        .await?;
+
+    task_handle.await?;
+
+    Ok(())
+}
diff --git a/example_apps/aggregations/src/main.rs b/example_apps/aggregations/src/main.rs
index 47bb322ad..3c1bb9f1e 100644
--- a/example_apps/aggregations/src/main.rs
+++ b/example_apps/aggregations/src/main.rs
@@ -49,6 +49,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let aggregate_agent = AgentModel::new(CityAgent::default, CityLifecycle.into_lifecycle());
 
     let mut builder = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(RoutePattern::parse_str("/cars/:car_id")?, car_agent)
         .add_route(RoutePattern::parse_str("/city")?, aggregate_agent)
         .update_config(|config| {
diff --git a/example_apps/command_lane/Cargo.toml b/example_apps/command_lane/Cargo.toml
index c54ddd9d6..9acfb6779 100644
--- a/example_apps/command_lane/Cargo.toml
+++ b/example_apps/command_lane/Cargo.toml
@@ -6,6 +6,10 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 swimos_form = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+
+[[bin]]
+name = "command_client"
diff --git a/example_apps/command_lane/src/bin/command_client.rs b/example_apps/command_lane/src/bin/command_client.rs
new file mode 100644
index 000000000..e95140cde
--- /dev/null
+++ b/example_apps/command_lane/src/bin/command_client.rs
@@ -0,0 +1,36 @@
+use command_lane::Instruction;
+use std::error::Error;
+use swimos_client::{BasicValueDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let state_lifecycle =
+        BasicValueDownlinkLifecycle::default().on_event_blocking(|state| println!("{state}"));
+    client_handle
+        .value_downlink::<i32>(RemotePath::new(host, "/example/1", "lane"))
+        .lifecycle(state_lifecycle)
+        .open()
+        .await?;
+
+    let command_handle = client_handle
+        .value_downlink::<Instruction>(RemotePath::new(host, "/example/1", "command"))
+        .open()
+        .await?;
+
+    for i in 0..10 {
+        command_handle.set(Instruction::Add(i)).await?;
+    }
+
+    command_handle.set(Instruction::Zero).await?;
+
+    task_handle.await?;
+
+    Ok(())
+}
diff --git a/example_apps/command_lane/src/lib.rs b/example_apps/command_lane/src/lib.rs
new file mode 100644
index 000000000..73881765d
--- /dev/null
+++ b/example_apps/command_lane/src/lib.rs
@@ -0,0 +1,3 @@
+mod model;
+
+pub use model::Instruction;
diff --git a/example_apps/command_lane/src/main.rs b/example_apps/command_lane/src/main.rs
index 6daaaad1f..e68489dcd 100644
--- a/example_apps/command_lane/src/main.rs
+++ b/example_apps/command_lane/src/main.rs
@@ -35,6 +35,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/example_apps/demand_lane/Cargo.toml b/example_apps/demand_lane/Cargo.toml
index fc7cca084..e595acc69 100644
--- a/example_apps/demand_lane/Cargo.toml
+++ b/example_apps/demand_lane/Cargo.toml
@@ -6,5 +6,9 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+
+[[bin]]
+name = "demand_client"
diff --git a/example_apps/demand_lane/src/agent.rs b/example_apps/demand_lane/src/agent.rs
index e52f571b7..173051a19 100644
--- a/example_apps/demand_lane/src/agent.rs
+++ b/example_apps/demand_lane/src/agent.rs
@@ -31,7 +31,7 @@ pub struct ExampleAgent {
 #[derive(Clone)]
 pub struct ExampleLifecycle;
 
-const CUE_PERIOD: Duration = Duration::from_secs(5);
+const CUE_PERIOD: Duration = Duration::from_secs(1);
 
 #[lifecycle(ExampleAgent)]
 impl ExampleLifecycle {
diff --git a/example_apps/demand_lane/src/bin/demand_client.rs b/example_apps/demand_lane/src/bin/demand_client.rs
new file mode 100644
index 000000000..e604767a3
--- /dev/null
+++ b/example_apps/demand_lane/src/bin/demand_client.rs
@@ -0,0 +1,34 @@
+use std::error::Error;
+use swimos_client::{BasicValueDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let state_lifecycle =
+        BasicValueDownlinkLifecycle::default().on_event_blocking(|state| println!("{state}"));
+    let state_view = client_handle
+        .value_downlink::<i32>(RemotePath::new(host, "/example/1", "lane"))
+        .lifecycle(state_lifecycle)
+        .open()
+        .await?;
+
+    let demand_lifecycle =
+        BasicValueDownlinkLifecycle::default().on_event_blocking(|state| println!("{state}"));
+    client_handle
+        .value_downlink::<i32>(RemotePath::new(host, "/example/1", "demand"))
+        .lifecycle(demand_lifecycle)
+        .open()
+        .await?;
+
+    state_view.set(13).await?;
+
+    task_handle.await?;
+
+    Ok(())
+}
diff --git a/example_apps/demand_lane/src/main.rs b/example_apps/demand_lane/src/main.rs
index 480925394..6c501f4f8 100644
--- a/example_apps/demand_lane/src/main.rs
+++ b/example_apps/demand_lane/src/main.rs
@@ -34,6 +34,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/example_apps/demand_map_lane/Cargo.toml b/example_apps/demand_map_lane/Cargo.toml
index a197cea00..474dcd07a 100644
--- a/example_apps/demand_map_lane/Cargo.toml
+++ b/example_apps/demand_map_lane/Cargo.toml
@@ -6,5 +6,9 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+
+[[bin]]
+name = "demand_map_client"
diff --git a/example_apps/demand_map_lane/src/bin/demand_map_client.rs b/example_apps/demand_map_lane/src/bin/demand_map_client.rs
new file mode 100644
index 000000000..4639035b3
--- /dev/null
+++ b/example_apps/demand_map_lane/src/bin/demand_map_client.rs
@@ -0,0 +1,33 @@
+use std::error::Error;
+use swimos_client::{BasicMapDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let command_view = client_handle
+        .value_downlink::<String>(RemotePath::new(host, "/example/1", "command"))
+        .open()
+        .await?;
+
+    let demand_lifecycle = BasicMapDownlinkLifecycle::default()
+        .on_update_blocking(|key, _map, _old_state, new_state| println!("{key} -> {new_state}"));
+    client_handle
+        .map_downlink::<String, i32>(RemotePath::new(host, "/example/1", "demand_map"))
+        .lifecycle(demand_lifecycle)
+        .open()
+        .await?;
+
+    command_view.set("Red".to_string()).await?;
+    command_view.set("Green".to_string()).await?;
+    command_view.set("Blue".to_string()).await?;
+
+    task_handle.await?;
+
+    Ok(())
+}
diff --git a/example_apps/demand_map_lane/src/main.rs b/example_apps/demand_map_lane/src/main.rs
index f582904b8..cbe4acb67 100644
--- a/example_apps/demand_map_lane/src/main.rs
+++ b/example_apps/demand_map_lane/src/main.rs
@@ -35,6 +35,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/example_apps/http_lane/Cargo.toml b/example_apps/http_lane/Cargo.toml
index c53d3f2e4..5886bba61 100644
--- a/example_apps/http_lane/Cargo.toml
+++ b/example_apps/http_lane/Cargo.toml
@@ -8,3 +8,7 @@ publish = false
 swimos = { workspace = true, features = ["server", "agent"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+reqwest = { workspace = true }
+
+[[bin]]
+name = "http_client"
\ No newline at end of file
diff --git a/example_apps/http_lane/src/bin/http_client.rs b/example_apps/http_lane/src/bin/http_client.rs
new file mode 100644
index 000000000..c70b15634
--- /dev/null
+++ b/example_apps/http_lane/src/bin/http_client.rs
@@ -0,0 +1,17 @@
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let lane = "http://127.0.0.1:8080/example/1?lane=http_lane";
+
+    let client = reqwest::Client::new();
+    client.put(lane).body("13").send().await?;
+
+    let response = client.get(lane).send().await?;
+    let body = response.bytes().await?;
+
+    let num = std::str::from_utf8(body.as_ref())?.parse::<i32>()?;
+    assert_eq!(13, num);
+
+    Ok(())
+}
diff --git a/example_apps/http_lane/src/main.rs b/example_apps/http_lane/src/main.rs
index 9551d950d..1adc1e94a 100644
--- a/example_apps/http_lane/src/main.rs
+++ b/example_apps/http_lane/src/main.rs
@@ -35,6 +35,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/example_apps/join_map/Cargo.toml b/example_apps/join_map/Cargo.toml
index c989a90a7..7d3c03648 100644
--- a/example_apps/join_map/Cargo.toml
+++ b/example_apps/join_map/Cargo.toml
@@ -6,5 +6,9 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
-example-util = { path = "../example_util" }
\ No newline at end of file
+example-util = { path = "../example_util" }
+
+[[bin]]
+name = "join_map_client"
diff --git a/example_apps/join_map/src/bin/join_map_client.rs b/example_apps/join_map/src/bin/join_map_client.rs
new file mode 100644
index 000000000..03a26882a
--- /dev/null
+++ b/example_apps/join_map/src/bin/join_map_client.rs
@@ -0,0 +1,43 @@
+use std::error::Error;
+
+use swimos_client::{BasicMapDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let aggregated_lifecycle = BasicMapDownlinkLifecycle::default()
+        .on_update_blocking(|key, _map, _old_state, new_state| println!("{key} -> {new_state}"));
+    client_handle
+        .map_downlink::<String, u64>(RemotePath::new(host, "/join/state/example", "streets"))
+        .lifecycle(aggregated_lifecycle)
+        .open()
+        .await?;
+
+    let california_view = client_handle
+        .map_downlink::<String, u64>(RemotePath::new(host, "/state/california", "state"))
+        .open()
+        .await?;
+    california_view.update("street_a".to_string(), 1).await?;
+
+    let texas_view = client_handle
+        .map_downlink::<String, u64>(RemotePath::new(host, "/state/texas", "state"))
+        .open()
+        .await?;
+    texas_view.update("street_b".to_string(), 2).await?;
+
+    let florida_view = client_handle
+        .map_downlink::<String, u64>(RemotePath::new(host, "/state/florida", "state"))
+        .open()
+        .await?;
+    florida_view.update("street_c".to_string(), 3).await?;
+
+    task_handle.await?;
+
+    Ok(())
+}
diff --git a/example_apps/join_map/src/main.rs b/example_apps/join_map/src/main.rs
index e511ca9f6..6ebb25c52 100644
--- a/example_apps/join_map/src/main.rs
+++ b/example_apps/join_map/src/main.rs
@@ -41,6 +41,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     );
 
     let server = ServerBuilder::with_plane_name("Statistics Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(street_route, street_agent)
         .add_route(aggregated_route, aggregated_agent)
         .update_config(|config| {
diff --git a/example_apps/join_value/Cargo.toml b/example_apps/join_value/Cargo.toml
index e344f1393..6bb67dedd 100644
--- a/example_apps/join_value/Cargo.toml
+++ b/example_apps/join_value/Cargo.toml
@@ -6,5 +6,10 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+rand = { workspace = true }
+
+[[bin]]
+name = "join_value_client"
\ No newline at end of file
diff --git a/example_apps/join_value/src/bin/join_value_client.rs b/example_apps/join_value/src/bin/join_value_client.rs
new file mode 100644
index 000000000..cebb587e5
--- /dev/null
+++ b/example_apps/join_value/src/bin/join_value_client.rs
@@ -0,0 +1,67 @@
+use rand::Rng;
+use std::error::Error;
+use std::time::Duration;
+use swimos_client::{
+    BasicMapDownlinkLifecycle, BasicValueDownlinkLifecycle, RemotePath, SwimClientBuilder,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let rooms = ["living_room", "kitchen", "office"];
+    let buildings = ["a", "b", "c"];
+
+    for building in buildings {
+        let aggregated_lifecycle = BasicMapDownlinkLifecycle::default().on_update_blocking(
+            |key, _map, _old_state, new_state| println!("Building {key} -> {new_state}"),
+        );
+        client_handle
+            .map_downlink::<String, bool>(RemotePath::new(
+                host,
+                format!("/buildings/{building}"),
+                "lights",
+            ))
+            .lifecycle(aggregated_lifecycle)
+            .open()
+            .await?;
+    }
+
+    for building in buildings {
+        for room in rooms {
+            let handle = client_handle.clone();
+            let task = async move {
+                let node_address = format!("/rooms/{building}/{building}_{room}");
+                let building_lifecycle = BasicValueDownlinkLifecycle::default()
+                    .on_synced_blocking(move |state| println!("Building '{building}' -> {state}"));
+                let building_view = handle
+                    .value_downlink::<bool>(RemotePath::new(host, node_address.as_str(), "lights"))
+                    .lifecycle(building_lifecycle)
+                    .open()
+                    .await
+                    .unwrap_or_else(|_| panic!("Failed to open downlink to {node_address}"));
+
+                let mut rng = rand::rngs::OsRng;
+                loop {
+                    let state = rng.gen_bool(0.5);
+                    building_view
+                        .set(state)
+                        .await
+                        .expect("Failed to set downlink");
+
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+                }
+            };
+            tokio::spawn(task);
+        }
+    }
+
+    task_handle.await?;
+
+    Ok(())
+}
diff --git a/example_apps/join_value/src/building.rs b/example_apps/join_value/src/building.rs
index d8b8b51b1..011a60c63 100644
--- a/example_apps/join_value/src/building.rs
+++ b/example_apps/join_value/src/building.rs
@@ -8,8 +8,8 @@ use swimos::agent::{lifecycle, AgentLaneModel};
 #[derive(AgentLaneModel)]
 #[projections]
 pub struct BuildingAgent {
-    lights: JoinValueLane<u64, bool>,
-    register_room: CommandLane<u64>,
+    lights: JoinValueLane<String, bool>,
+    register_room: CommandLane<String>,
 }
 
 #[derive(Clone)]
@@ -41,30 +41,32 @@ impl BuildingLifecycle {
     pub fn register_room(
         &self,
         context: HandlerContext<BuildingAgent>,
-        room_id: &u64,
+        room_id: &str,
     ) -> impl EventHandler<BuildingAgent> {
-        let room_id = *room_id;
+        let downlink_room_id = room_id.to_string();
+        let handler_room_id = room_id.to_string();
         context
             .get_parameter("name")
             .and_then(move |building_name: Option<String>| {
                 let building_name = building_name.expect("Missing building name URI parameter");
+                let node = format!("/rooms/{building_name}/{downlink_room_id}");
                 context.add_downlink(
                     BuildingAgent::LIGHTS,
-                    room_id,
+                    downlink_room_id,
                     None,
-                    format!("/rooms/{building_name}/{room_id}").as_str(),
+                    node.as_str(),
                     "lights",
                 )
             })
-            .followed_by(context.effect(move || println!("Registered room: {room_id}")))
+            .followed_by(context.effect(move || println!("Registered room: {handler_room_id}")))
     }
 
     #[on_update(lights)]
     fn lights(
         &self,
         context: HandlerContext<BuildingAgent>,
-        _map: &HashMap<u64, bool>,
-        key: u64,
+        _map: &HashMap<String, bool>,
+        key: String,
         prev: Option<bool>,
         new_value: &bool,
     ) -> impl EventHandler<BuildingAgent> {
diff --git a/example_apps/join_value/src/main.rs b/example_apps/join_value/src/main.rs
index d22b9462c..199bc23e0 100644
--- a/example_apps/join_value/src/main.rs
+++ b/example_apps/join_value/src/main.rs
@@ -38,6 +38,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let room_agent = AgentModel::new(RoomAgent::default, RoomLifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Building Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(building_route, building_agent)
         .add_route(room_route, room_agent)
         .update_config(|config| {
diff --git a/example_apps/join_value/src/room.rs b/example_apps/join_value/src/room.rs
index 73ec4d7c5..078da9ea0 100644
--- a/example_apps/join_value/src/room.rs
+++ b/example_apps/join_value/src/room.rs
@@ -1,4 +1,3 @@
-use std::str::FromStr;
 use swimos::agent::agent_lifecycle::HandlerContext;
 use swimos::agent::event_handler::{EventHandler, HandlerActionExt};
 use swimos::agent::lanes::ValueLane;
@@ -29,14 +28,12 @@ impl RoomLifecycle {
                             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,
+                                room_id_str,
                             )
                         })
                 },
diff --git a/example_apps/map_lane/Cargo.toml b/example_apps/map_lane/Cargo.toml
index 9ce2a2b0c..68bd121f3 100644
--- a/example_apps/map_lane/Cargo.toml
+++ b/example_apps/map_lane/Cargo.toml
@@ -6,5 +6,9 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+
+[[bin]]
+name = "map_client"
\ No newline at end of file
diff --git a/example_apps/map_lane/src/bin/map_client.rs b/example_apps/map_lane/src/bin/map_client.rs
new file mode 100644
index 000000000..17c73c053
--- /dev/null
+++ b/example_apps/map_lane/src/bin/map_client.rs
@@ -0,0 +1,31 @@
+use std::error::Error;
+use swimos_client::{BasicMapDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let state_lifecycle = BasicMapDownlinkLifecycle::default().on_update_blocking(
+        |key, _map, old_value, new_value| {
+            println!("'{key:?}' changed from '{old_value:?}' to '{new_value:?}'")
+        },
+    );
+    let state_view = client_handle
+        .map_downlink(RemotePath::new(host, "/example/1", "lane"))
+        .lifecycle(state_lifecycle)
+        .open()
+        .await?;
+
+    let keys = [(1, "a"), (2, "b"), (3, "c")];
+
+    for (key, value) in keys {
+        state_view.update(key, value.to_string()).await?;
+    }
+
+    task_handle.await?;
+    Ok(())
+}
diff --git a/example_apps/map_lane/src/main.rs b/example_apps/map_lane/src/main.rs
index 480925394..6c501f4f8 100644
--- a/example_apps/map_lane/src/main.rs
+++ b/example_apps/map_lane/src/main.rs
@@ -34,6 +34,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/example_apps/supply_lane/Cargo.toml b/example_apps/supply_lane/Cargo.toml
index 85e868f8a..721f1de91 100644
--- a/example_apps/supply_lane/Cargo.toml
+++ b/example_apps/supply_lane/Cargo.toml
@@ -6,5 +6,9 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+
+[[bin]]
+name = "supply_client"
\ No newline at end of file
diff --git a/example_apps/supply_lane/src/bin/supply_client.rs b/example_apps/supply_lane/src/bin/supply_client.rs
new file mode 100644
index 000000000..87c21643e
--- /dev/null
+++ b/example_apps/supply_lane/src/bin/supply_client.rs
@@ -0,0 +1,34 @@
+use std::error::Error;
+use swimos_client::{BasicEventDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let state_view = client_handle
+        .value_downlink(RemotePath::new(host, "/example/1", "lane"))
+        .open()
+        .await?;
+
+    let doubled_lifecycle = BasicEventDownlinkLifecycle::default().on_event_blocking(|event| {
+        println!("{event:?}");
+    });
+    client_handle
+        .event_downlink::<i32>(RemotePath::new(host, "/example/1", "supply"))
+        .lifecycle(doubled_lifecycle)
+        .open()
+        .await?;
+
+    let values = [1, 2, 3, 4, 5];
+
+    for value in values {
+        state_view.set(value).await?;
+    }
+
+    task_handle.await?;
+    Ok(())
+}
diff --git a/example_apps/supply_lane/src/main.rs b/example_apps/supply_lane/src/main.rs
index 480925394..6c501f4f8 100644
--- a/example_apps/supply_lane/src/main.rs
+++ b/example_apps/supply_lane/src/main.rs
@@ -34,6 +34,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/example_apps/value_lane/Cargo.toml b/example_apps/value_lane/Cargo.toml
index 820aa7077..83d34cffc 100644
--- a/example_apps/value_lane/Cargo.toml
+++ b/example_apps/value_lane/Cargo.toml
@@ -6,5 +6,9 @@ publish = false
 
 [dependencies]
 swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
+
+[[bin]]
+name = "value_client"
\ No newline at end of file
diff --git a/example_apps/value_lane/src/bin/value_client.rs b/example_apps/value_lane/src/bin/value_client.rs
new file mode 100644
index 000000000..f85ace115
--- /dev/null
+++ b/example_apps/value_lane/src/bin/value_client.rs
@@ -0,0 +1,30 @@
+use std::error::Error;
+use swimos_client::{BasicValueDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let lifecycle =
+        BasicValueDownlinkLifecycle::default().on_set_blocking(|old_value, new_value| {
+            println!("'{old_value:?}' -> '{new_value:?}'");
+        });
+    let view = client_handle
+        .value_downlink::<i32>(RemotePath::new(host, "/example/1", "lane"))
+        .lifecycle(lifecycle)
+        .open()
+        .await?;
+
+    let values = [1, 2, 3, 4, 5];
+
+    for value in values {
+        view.set(value).await?;
+    }
+
+    task_handle.await?;
+    Ok(())
+}
diff --git a/example_apps/value_lane/src/main.rs b/example_apps/value_lane/src/main.rs
index 9551d950d..1adc1e94a 100644
--- a/example_apps/value_lane/src/main.rs
+++ b/example_apps/value_lane/src/main.rs
@@ -35,6 +35,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let agent = AgentModel::new(ExampleAgent::default, lifecycle.into_lifecycle());
 
     let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
         .add_route(route, agent)
         .update_config(|config| {
             config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
diff --git a/runtime/swimos_runtime/src/downlink/mod.rs b/runtime/swimos_runtime/src/downlink/mod.rs
index f63d9abcb..f58cb5cf6 100644
--- a/runtime/swimos_runtime/src/downlink/mod.rs
+++ b/runtime/swimos_runtime/src/downlink/mod.rs
@@ -13,6 +13,8 @@
 // limitations under the License.
 
 use std::collections::HashSet;
+use std::error::Error;
+use std::fmt::Debug;
 use std::num::NonZeroUsize;
 use std::time::Duration;
 
@@ -510,9 +512,12 @@ impl<D: Decoder> DownlinkReceiver<D> {
     }
 }
 
-struct Failed(u64);
+struct Failed(u64, Box<dyn Error + 'static>);
 
-impl<D: Decoder> Stream for DownlinkReceiver<D> {
+impl<D: Decoder> Stream for DownlinkReceiver<D>
+where
+    D::Error: Error + 'static,
+{
     type Item = Result<D::Item, Failed>;
 
     fn poll_next(
@@ -525,7 +530,7 @@ impl<D: Decoder> Stream for DownlinkReceiver<D> {
         } else {
             this.receiver
                 .poll_next_unpin(cx)
-                .map_err(|_| Failed(this.id))
+                .map_err(|e| Failed(this.id, Box::new(e)))
         }
     }
 }
@@ -958,7 +963,9 @@ async fn write_task<B: DownlinkBackpressure>(
     config: DownlinkRuntimeConfig,
     mut backpressure: B,
     stop_voter: Voter,
-) {
+) where
+    <<B as DownlinkBackpressure>::Dec as Decoder>::Error: Error + 'static,
+{
     let mut message_writer = RequestSender::new(output, identity, path);
     if message_writer.send_link().await.is_err() {
         return;
@@ -1173,8 +1180,8 @@ async fn write_task<B: DownlinkBackpressure>(
                             }
                         },
                         Either::Right(ow) => {
-                            if let Some(Err(Failed(id))) = ow {
-                                trace!("Removing a failed subscriber");
+                            if let Some(Err(Failed(id, error))) = ow {
+                                trace!(?error, "Removing a failed subscriber");
                                 if let Some(rx) = registered.iter_mut().find(|rx| rx.id == id) {
                                     rx.terminate();
                                 }
@@ -1249,11 +1256,14 @@ async fn write_task<B: DownlinkBackpressure>(
                             // Writing is currently blocked so overwrite the next value to be sent.
                             trace!("Over-writing the current event buffer.");
                             if let Err(err) = backpressure.push_operation(op) {
-                                error!("Failed to process downlink operaton: {error}", error = err);
+                                error!(
+                                    "Failed to process downlink operation: {error}",
+                                    error = err
+                                );
                             };
                         }
-                        SuspendedResult::NextRecord(Some(Err(Failed(id)))) => {
-                            trace!("Removing a failed subscriber.");
+                        SuspendedResult::NextRecord(Some(Err(Failed(id, error)))) => {
+                            trace!(?error, "Removing a failed subscriber.");
                             if let Some(rx) = registered.iter_mut().find(|rx| rx.id == id) {
                                 rx.terminate();
                             }
diff --git a/swimos_client/src/lib.rs b/swimos_client/src/lib.rs
index c909825a7..469468c66 100644
--- a/swimos_client/src/lib.rs
+++ b/swimos_client/src/lib.rs
@@ -375,6 +375,7 @@ impl<'h, L> ValueDownlinkBuilder<'h, L> {
     /// Sets a new lifecycle that to be used.
     pub fn lifecycle<NL, T>(self, lifecycle: NL) -> ValueDownlinkBuilder<'h, NL>
     where
+        L: ValueDownlinkLifecycle<T>,
         NL: ValueDownlinkLifecycle<T>,
     {
         let ValueDownlinkBuilder {
@@ -442,9 +443,11 @@ impl<'h, L> ValueDownlinkBuilder<'h, L> {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, thiserror::Error)]
 pub enum ValueDownlinkOperationError {
+    #[error("Downlink has not yet synced")]
     NotYetSynced,
+    #[error("Downlink terminated")]
     DownlinkStopped,
 }
 
@@ -500,6 +503,7 @@ impl<'h, L> MapDownlinkBuilder<'h, L> {
     /// Sets a new lifecycle that to be used.
     pub fn lifecycle<K, V, NL>(self, lifecycle: NL) -> MapDownlinkBuilder<'h, NL>
     where
+        L: MapDownlinkLifecycle<K, V>,
         NL: MapDownlinkLifecycle<K, V>,
     {
         let MapDownlinkBuilder {
@@ -595,16 +599,6 @@ impl<K, V> MapDownlinkView<K, V> {
         self.inner.clear().await
     }
 
-    /// Retains the last `n` elements in the map.
-    pub async fn take(&self, n: u64) -> Result<(), ChannelError> {
-        self.inner.take(n).await
-    }
-
-    /// Retains the first `n` elements in the map.
-    pub async fn drop(&self, n: u64) -> Result<(), ChannelError> {
-        self.inner.drop(n).await
-    }
-
     /// Returns a receiver that completes with the result of downlink's internal task.
     pub fn stop_notification(&self) -> promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>> {
         self.stop_rx.clone()
@@ -625,6 +619,7 @@ impl<'h, L> EventDownlinkBuilder<'h, L> {
     /// Sets a new lifecycle that to be used.
     pub fn lifecycle<T, NL>(self, lifecycle: NL) -> EventDownlinkBuilder<'h, NL>
     where
+        L: EventDownlinkLifecycle<T>,
         NL: EventDownlinkLifecycle<T>,
     {
         let EventDownlinkBuilder {
diff --git a/swimos_client/src/runtime.rs b/swimos_client/src/runtime.rs
index afd360782..7ee8a4725 100644
--- a/swimos_client/src/runtime.rs
+++ b/swimos_client/src/runtime.rs
@@ -318,6 +318,8 @@ async fn runtime_task<Net, Ws, Provider>(
                     runtime_config,
                     downlink_config,
                 } = request;
+                trace!(%path, "Received downlink registration request");
+
                 let RemotePath { host, node, lane } = path;
                 let shp = match host.as_str().parse::<SchemeHostPort>() {
                     Ok(shp) => shp,
@@ -361,6 +363,7 @@ async fn runtime_task<Net, Ws, Provider>(
                 host,
                 result: Ok(addrs),
             } => {
+                trace!(?scheme, ?host, "Resolved host");
                 match addrs
                     .iter()
                     .find_map(|sock| peers.get(sock).map(|peer| (peer, sock)))
@@ -504,6 +507,8 @@ async fn runtime_task<Net, Ws, Provider>(
                 result: Ok((attach, runtime)),
             } => match peers.get_mut(&sock) {
                 Some(peer) => {
+                    debug_assert!(peer.get_view(&key).is_none());
+                    trace!(?key, "Downlink runtime started");
                     let (runtime_stop_tx, runtime_stop_rx) = trigger::trigger();
                     let peer_key = key.clone();
                     downlinks.push(
@@ -576,6 +581,7 @@ async fn runtime_task<Net, Ws, Provider>(
                     downlink_config,
                     ..
                 } = pending;
+                trace!(?address, "Downlink runtime attached");
                 let kind = downlink.kind();
                 let (promise_tx, promise_rx) = promise::promise();
                 let task = tokio::spawn(downlink.run_boxed(
@@ -643,6 +649,7 @@ async fn runtime_task<Net, Ws, Provider>(
                 result,
                 tx,
             } => {
+                trace!(?address, "Downlink runtime complete");
                 match &result {
                     Ok(()) => {
                         info!(
@@ -671,9 +678,8 @@ async fn runtime_task<Net, Ws, Provider>(
                 break;
             }
         }
-
-        debug!("Runtime task completed");
     }
+    debug!("Runtime task completed");
 }
 
 async fn start_downlink_runtime(
@@ -689,6 +695,8 @@ async fn start_downlink_runtime(
     let (out_tx, out_rx) = byte_channel(config.remote_buffer_size);
     let (done_tx, done_rx) = oneshot::channel();
 
+    trace!(?rel_addr, "Resolving attach handle");
+
     let request = AttachClient::AttachDownlink {
         downlink_id: identity,
         path: rel_addr.clone(),
diff --git a/swimos_downlink/src/model/mod.rs b/swimos_downlink/src/model/mod.rs
index f26779c17..c99709fcf 100644
--- a/swimos_downlink/src/model/mod.rs
+++ b/swimos_downlink/src/model/mod.rs
@@ -16,7 +16,7 @@ use std::error::Error;
 use std::fmt::{Display, Formatter};
 use std::marker::PhantomData;
 
-use swimos_agent_protocol::MapMessage;
+use swimos_agent_protocol::MapOperation;
 use tokio::sync::{mpsc, oneshot};
 
 use crate::model::lifecycle::{BasicMapDownlinkLifecycle, MapDownlinkLifecycle};
@@ -62,13 +62,13 @@ impl<T, LC> EventDownlinkModel<T, LC> {
 }
 
 pub struct MapDownlinkModel<K, V, LC> {
-    pub actions: mpsc::Receiver<MapMessage<K, V>>,
+    pub actions: mpsc::Receiver<MapOperation<K, V>>,
     pub lifecycle: LC,
 }
 
 impl<K, V, LC> MapDownlinkModel<K, V, LC> {
     pub fn new(
-        actions: mpsc::Receiver<MapMessage<K, V>>,
+        actions: mpsc::Receiver<MapOperation<K, V>>,
         lifecycle: LC,
     ) -> MapDownlinkModel<K, V, LC> {
         MapDownlinkModel { actions, lifecycle }
@@ -98,7 +98,7 @@ pub fn event_downlink<T>() -> DefaultEventDownlinkModel<T> {
 }
 
 pub fn map_downlink<K, V>(
-    actions: mpsc::Receiver<MapMessage<K, V>>,
+    actions: mpsc::Receiver<MapOperation<K, V>>,
 ) -> DefaultMapDownlinkModel<K, V> {
     MapDownlinkModel::new(actions, Default::default())
 }
@@ -129,37 +129,27 @@ impl From<oneshot::error::RecvError> for ChannelError {
 /// A map downlink handle.
 #[derive(Debug, Clone)]
 pub struct MapDownlinkHandle<K, V> {
-    inner: mpsc::Sender<MapMessage<K, V>>,
+    inner: mpsc::Sender<MapOperation<K, V>>,
 }
 
 impl<K, V> MapDownlinkHandle<K, V> {
-    pub fn new(inner: mpsc::Sender<MapMessage<K, V>>) -> MapDownlinkHandle<K, V> {
+    pub fn new(inner: mpsc::Sender<MapOperation<K, V>>) -> MapDownlinkHandle<K, V> {
         MapDownlinkHandle { inner }
     }
 
     /// Updates or inserts the key-value pair into the map.
     pub async fn update(&self, key: K, value: V) -> Result<(), ChannelError> {
-        Ok(self.inner.send(MapMessage::Update { key, value }).await?)
+        Ok(self.inner.send(MapOperation::Update { key, value }).await?)
     }
 
     /// Removes the value corresponding to the key.
     pub async fn remove(&self, key: K) -> Result<(), ChannelError> {
-        Ok(self.inner.send(MapMessage::Remove { key }).await?)
+        Ok(self.inner.send(MapOperation::Remove { key }).await?)
     }
 
     /// Clears the map, removing all of the elements.
     pub async fn clear(&self) -> Result<(), ChannelError> {
-        Ok(self.inner.send(MapMessage::Clear).await?)
-    }
-
-    /// Retains the last `n` elements in the map.
-    pub async fn take(&self, n: u64) -> Result<(), ChannelError> {
-        Ok(self.inner.send(MapMessage::Take(n)).await?)
-    }
-
-    /// Retains the first `n` elements in the map.
-    pub async fn drop(&self, n: u64) -> Result<(), ChannelError> {
-        Ok(self.inner.send(MapMessage::Drop(n)).await?)
+        Ok(self.inner.send(MapOperation::Clear).await?)
     }
 
     /// Completes when the downlink closes; a downlink closes when the connection closes or an
diff --git a/swimos_downlink/src/task/map.rs b/swimos_downlink/src/task/map.rs
index d4336180b..190921363 100644
--- a/swimos_downlink/src/task/map.rs
+++ b/swimos_downlink/src/task/map.rs
@@ -19,9 +19,10 @@ use futures::{FutureExt, Sink, SinkExt, StreamExt};
 use std::collections::BTreeMap;
 use std::fmt::{Debug, Display};
 use std::mem;
-use swimos_agent_protocol::encoding::downlink::{DownlinkOperationEncoder, MapNotificationDecoder};
-use swimos_agent_protocol::MapMessage;
-use swimos_agent_protocol::{DownlinkNotification, DownlinkOperation};
+use swimos_agent_protocol::encoding::downlink::MapNotificationDecoder;
+use swimos_agent_protocol::encoding::map::MapOperationEncoder;
+use swimos_agent_protocol::DownlinkNotification;
+use swimos_agent_protocol::{MapMessage, MapOperation};
 use swimos_api::{address::Address, error::DownlinkTaskError};
 use swimos_client_api::DownlinkConfig;
 use swimos_form::write::StructuralWritable;
@@ -63,7 +64,7 @@ where
         config,
         input,
         lifecycle,
-        FramedWrite::new(output, DownlinkOperationEncoder::default()),
+        FramedWrite::new(output, MapOperationEncoder),
         actions,
         MapNotificationDecoder::default(),
     )
@@ -97,7 +98,7 @@ where
 
 enum IoEvent<K, V> {
     Read(DownlinkNotification<MapMessage<K, V>>),
-    Write(Option<MapMessage<K, V>>),
+    Write(Option<MapOperation<K, V>>),
 }
 
 /// The current IO mode. Defaults to read/write and once the writer channel is dropped, the IO loop
@@ -112,7 +113,7 @@ async fn run_io<K, V, LC, Snk, D, E>(
     input: ByteReader,
     mut lifecycle: LC,
     mut framed: Snk,
-    actions: mpsc::Receiver<MapMessage<K, V>>,
+    actions: mpsc::Receiver<MapOperation<K, V>>,
     decoder: D,
 ) -> Result<(), DownlinkTaskError>
 where
@@ -120,7 +121,7 @@ where
     V: MapValue,
     V::Rec: Send,
     LC: MapDownlinkLifecycle<K, V>,
-    Snk: Sink<DownlinkOperation<MapMessage<K, V>>> + Unpin,
+    Snk: Sink<MapOperation<K, V>> + Unpin,
     Snk::Error: Debug,
     D: Decoder<Item = DownlinkNotification<MapMessage<K, V>>, Error = E>,
     DownlinkTaskError: From<E>,
@@ -151,31 +152,18 @@ where
 
                         match &mut state {
                             State::Synced(map) | State::Linked(map) => match &message {
-                                MapMessage::Update { key, value } => {
+                                MapOperation::Update { key, value } => {
                                     map.insert(K::clone(key), V::clone(value));
                                 }
-                                MapMessage::Remove { key } => {
+                                MapOperation::Remove { key } => {
                                     map.remove(key);
                                 }
-                                MapMessage::Clear => map.clear(),
-                                MapMessage::Take(cnt) => {
-                                    let mut it = mem::take(map).into_iter();
-                                    for (key, value) in (&mut it).take(*cnt as usize) {
-                                        map.insert(key, value);
-                                    }
-                                }
-                                MapMessage::Drop(cnt) => {
-                                    let it = mem::take(map).into_iter().skip(*cnt as usize);
-                                    for (key, value) in it {
-                                        map.insert(key, value);
-                                    }
-                                }
+                                MapOperation::Clear => map.clear(),
                             },
                             State::Unlinked => {}
                         }
 
-                        let op = DownlinkOperation::new(message);
-                        if let Err(e) = framed.feed(op).await {
+                        if let Err(e) = framed.feed(message).await {
                             error!(error = ?e, "Failed to feed downlink frame. Transitioning to read-only mode");
                         }
                     }
diff --git a/swimos_downlink/src/task/mod.rs b/swimos_downlink/src/task/mod.rs
index 1ac170bf8..227b113de 100644
--- a/swimos_downlink/src/task/mod.rs
+++ b/swimos_downlink/src/task/mod.rs
@@ -119,6 +119,7 @@ pub trait MapKey:
     Clone + Form + Send + Sync + Eq + Hash + Ord + RecognizerReadable + 'static
 {
 }
+
 impl<T> MapKey for T where
     T: Clone + Form + Send + Sync + Eq + Hash + Ord + RecognizerReadable + 'static
 {


Test patch:
diff --git a/swimos_downlink/src/task/tests/map.rs b/swimos_downlink/src/task/tests/map.rs
index 63ba34db8..dd8fd5cfe 100644
--- a/swimos_downlink/src/task/tests/map.rs
+++ b/swimos_downlink/src/task/tests/map.rs
@@ -18,11 +18,11 @@ use std::collections::BTreeMap;
 use std::future::Future;
 use std::hash::Hash;
 use std::num::NonZeroUsize;
-use swimos_agent_protocol::MapMessage;
+use swimos_agent_protocol::{MapMessage, MapOperation};
 use swimos_client_api::{Downlink, DownlinkConfig};
 
 use swimos_agent_protocol::encoding::downlink::DownlinkNotificationEncoder;
-use swimos_agent_protocol::encoding::map::MapMessageEncoder;
+use swimos_agent_protocol::encoding::map::{MapMessageEncoder, MapOperationDecoder};
 use swimos_agent_protocol::DownlinkNotification;
 use swimos_api::error::{DownlinkTaskError, FrameIoError, InvalidFrame};
 use swimos_form::write::StructuralWritable;
@@ -558,12 +558,12 @@ async fn send_on_downlink() {
         |writer, mut reader| async move {
             let _writer = writer;
             assert!(set_tx
-                .send(MapMessage::Update { key: 1, value: 1 })
+                .send(MapOperation::Update { key: 1, value: 1 })
                 .await
                 .is_ok());
             assert_eq!(
-                reader.recv::<MapMessage<i32, i32>>().await,
-                Ok(Some(MapMessage::Update { key: 1, value: 1 }))
+                reader.decode(MapOperationDecoder::default()).await,
+                Ok(Some(MapOperation::Update { key: 1, value: 1 }))
             );
         },
     )
@@ -718,58 +718,6 @@ async fn rx_take_elem_downlink() {
     assert!(result.unwrap().recv().await.is_none());
 }
 
-#[tokio::test]
-async fn handle_take_elem_downlink() {
-    let (event_tx, mut event_rx) = mpsc::unbounded_channel::<TestMessage<i32, i32>>();
-    let (set_tx, set_rx) = mpsc::channel(16);
-    let handle = MapDownlinkHandle::new(set_tx);
-    let lifecycle = make_lifecycle(event_tx);
-    let model = MapDownlinkModel::new(set_rx, lifecycle);
-
-    let config = DownlinkConfig {
-        events_when_not_synced: false,
-        terminate_on_unlinked: true,
-        buffer_size: DEFAULT_BUFFER_SIZE,
-    };
-
-    let result = run_map_downlink_task(
-        DownlinkTask::new(model),
-        config,
-        |mut writer, mut reader| async move {
-            writer
-                .send_message::<i32, i32>(DownlinkNotification::Linked)
-                .await;
-
-            for i in 0..5 {
-                writer
-                    .send_message::<i32, i32>(DownlinkNotification::Event {
-                        body: MapMessage::Update { key: i, value: i },
-                    })
-                    .await;
-            }
-
-            writer
-                .send_message::<i32, i32>(DownlinkNotification::Synced)
-                .await;
-            expect_event(&mut event_rx, TestMessage::Linked).await;
-
-            let state = (0..5).map(|i| (i, i)).collect::<BTreeMap<i32, i32>>();
-            expect_event(&mut event_rx, TestMessage::Synced(state)).await;
-
-            assert!(handle.take(2).await.is_ok());
-            assert_eq!(
-                reader.recv::<MapMessage<i32, i32>>().await,
-                Ok(Some(MapMessage::Take(2)))
-            );
-
-            event_rx
-        },
-    )
-    .await;
-    assert!(result.is_ok());
-    assert!(result.unwrap().recv().await.is_none());
-}
-
 #[tokio::test]
 async fn rx_drop_elem_downlink() {
     let (event_tx, mut event_rx) = mpsc::unbounded_channel::<TestMessage<i32, i32>>();
@@ -833,59 +781,6 @@ async fn rx_drop_elem_downlink() {
     assert!(result.unwrap().recv().await.is_none());
 }
 
-#[tokio::test]
-async fn handle_drop_elem_downlink() {
-    let (event_tx, mut event_rx) = mpsc::unbounded_channel::<TestMessage<i32, i32>>();
-    let (set_tx, set_rx) = mpsc::channel(16);
-    let handle = MapDownlinkHandle::new(set_tx);
-    let lifecycle = make_lifecycle(event_tx);
-    let model = MapDownlinkModel::new(set_rx, lifecycle);
-
-    let config = DownlinkConfig {
-        events_when_not_synced: false,
-        terminate_on_unlinked: true,
-        buffer_size: DEFAULT_BUFFER_SIZE,
-    };
-
-    let result = run_map_downlink_task(
-        DownlinkTask::new(model),
-        config,
-        |mut writer, mut reader| async move {
-            writer
-                .send_message::<i32, i32>(DownlinkNotification::Linked)
-                .await;
-
-            for i in 0..5 {
-                writer
-                    .send_message::<i32, i32>(DownlinkNotification::Event {
-                        body: MapMessage::Update { key: i, value: i },
-                    })
-                    .await;
-            }
-
-            writer
-                .send_message::<i32, i32>(DownlinkNotification::Synced)
-                .await;
-            expect_event(&mut event_rx, TestMessage::Linked).await;
-
-            let state = (0..5).map(|i| (i, i)).collect::<BTreeMap<i32, i32>>();
-            expect_event(&mut event_rx, TestMessage::Synced(state)).await;
-
-            assert!(handle.drop(2).await.is_ok());
-
-            assert_eq!(
-                reader.recv::<MapMessage<i32, i32>>().await,
-                Ok(Some(MapMessage::Drop(2)))
-            );
-
-            event_rx
-        },
-    )
-    .await;
-    assert!(result.is_ok());
-    assert!(result.unwrap().recv().await.is_none());
-}
-
 #[tokio::test]
 async fn remove_elem_downlink() {
     let (event_tx, mut event_rx) = mpsc::unbounded_channel::<TestMessage<i32, i32>>();
diff --git a/swimos_downlink/src/task/tests/mod.rs b/swimos_downlink/src/task/tests/mod.rs
index bed3019f2..97889e769 100644
--- a/swimos_downlink/src/task/tests/mod.rs
+++ b/swimos_downlink/src/task/tests/mod.rs
@@ -17,17 +17,14 @@ use std::num::NonZeroUsize;
 
 use futures::future::join;
 use futures::{SinkExt, StreamExt};
-use swimos_client_api::{Downlink, DownlinkConfig};
 use tokio::time::{timeout, Duration};
-use tokio_util::codec::{FramedRead, FramedWrite};
+use tokio_util::codec::{Decoder, FramedRead, FramedWrite};
 
-use swimos_agent_protocol::encoding::downlink::{
-    DownlinkNotificationEncoder, DownlinkOperationDecoder,
-};
-use swimos_agent_protocol::{DownlinkNotification, DownlinkOperation};
+use swimos_agent_protocol::encoding::downlink::DownlinkNotificationEncoder;
+use swimos_agent_protocol::DownlinkNotification;
 use swimos_api::{address::Address, error::DownlinkTaskError};
-use swimos_form::{read::RecognizerReadable, write::StructuralWritable};
-use swimos_recon::parser::parse_recognize;
+use swimos_client_api::{Downlink, DownlinkConfig};
+use swimos_form::write::StructuralWritable;
 use swimos_recon::print_recon_compact;
 use swimos_utilities::{
     byte_channel::{byte_channel, ByteReader, ByteWriter},
@@ -49,11 +46,12 @@ impl TestValueWriter {
     }
 }
 
-struct TestReader(FramedRead<ByteReader, DownlinkOperationDecoder>);
+//FramedRead<ByteReader, DownlinkOperationDecoder>
+struct TestReader(ByteReader);
 
 impl TestReader {
     fn new(rx: ByteReader) -> Self {
-        TestReader(FramedRead::new(rx, DownlinkOperationDecoder))
+        TestReader(rx)
     }
 }
 
@@ -73,19 +71,13 @@ impl From<std::str::Utf8Error> for ReadFailed {
 }
 
 impl TestReader {
-    async fn recv<T: RecognizerReadable>(&mut self) -> Result<Option<T>, ReadFailed> {
+    async fn decode<D>(&mut self, dec: D) -> Result<Option<D::Item>, ReadFailed>
+    where
+        D: Decoder,
+    {
         let TestReader(inner) = self;
-        let op = inner.next().await.transpose()?;
-        if let Some(DownlinkOperation { body }) = op {
-            let body_str = std::str::from_utf8(body.as_ref())?;
-            if let Ok(v) = parse_recognize(body_str, false) {
-                Ok(Some(v))
-            } else {
-                Err(ReadFailed)
-            }
-        } else {
-            Ok(None)
-        }
+        let mut dec = FramedRead::new(inner, dec);
+        dec.next().await.transpose().map_err(|_| ReadFailed)
     }
 }
 
diff --git a/swimos_downlink/src/task/tests/value.rs b/swimos_downlink/src/task/tests/value.rs
index 293a3549c..b495f308d 100644
--- a/swimos_downlink/src/task/tests/value.rs
+++ b/swimos_downlink/src/task/tests/value.rs
@@ -13,20 +13,27 @@
 // limitations under the License.
 
 use std::fmt::Debug;
+use std::marker::PhantomData;
 use std::num::NonZeroUsize;
 
+use bytes::BytesMut;
+use tokio::sync::mpsc;
+use tokio_util::codec::Decoder;
+
+use swimos_agent_protocol::encoding::downlink::DownlinkOperationDecoder;
 use swimos_agent_protocol::DownlinkNotification;
 use swimos_api::error::{DownlinkTaskError, FrameIoError, InvalidFrame};
-
 use swimos_client_api::DownlinkConfig;
+use swimos_form::read::RecognizerReadable;
+use swimos_recon::parser::parse_recognize;
 use swimos_utilities::non_zero_usize;
-use tokio::sync::mpsc;
 
-use super::run_value_downlink_task;
 use crate::model::lifecycle::{BasicValueDownlinkLifecycle, ValueDownlinkLifecycle};
 use crate::model::ValueDownlinkSet;
 use crate::{DownlinkTask, ValueDownlinkModel};
 
+use super::run_value_downlink_task;
+
 #[derive(Debug, PartialEq, Eq)]
 enum TestMessage<T> {
     Linked,
@@ -393,6 +400,36 @@ async fn relink_downlink() {
     assert!(result.unwrap().recv().await.is_none());
 }
 
+struct StringDecoder<T>(PhantomData<T>);
+
+impl<T> Default for StringDecoder<T> {
+    fn default() -> Self {
+        StringDecoder(PhantomData)
+    }
+}
+
+impl<T> Decoder for StringDecoder<T>
+where
+    T: RecognizerReadable,
+{
+    type Item = T;
+    type Error = std::io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        let err = || std::io::ErrorKind::Other.into();
+        match DownlinkOperationDecoder.decode(src)? {
+            Some(op) => {
+                let body_str = std::str::from_utf8(op.body.as_ref()).map_err(|_| err())?;
+                match parse_recognize(body_str, false) {
+                    Ok(v) => Ok(Some(v)),
+                    Err(_) => Err(err()),
+                }
+            }
+            None => Ok(None),
+        }
+    }
+}
+
 #[tokio::test]
 async fn send_on_downlink() {
     let (event_tx, _event_rx) = mpsc::unbounded_channel::<TestMessage<i32>>();
@@ -412,7 +449,7 @@ async fn send_on_downlink() {
         |writer, mut reader| async move {
             let _writer = writer;
             handle_tx.send(ValueDownlinkSet { to: 12 }).await.unwrap();
-            assert_eq!(reader.recv::<i32>().await, Ok(Some(12)));
+            assert_eq!(reader.decode(StringDecoder::default()).await, Ok(Some(12)));
         },
     )
     .await;