Resolves PR comments

SirCipher · SirCipher · commit 6bc7f7254e11 · 2024-07-11T15:48:41.000+01:00
diff --git a/example_apps/aggregations/Cargo.toml b/example_apps/aggregations/Cargo.toml
@@ -13,5 +13,5 @@ example-util = { path = "../example_util" }
 rand = { workspace = true }
 futures-util = { workspace = true }
 
-[[example]]
+[[bin]]
 name = "aggregations_client"
diff --git a/example_apps/aggregations/src/bin/aggregations_client.rs b/example_apps/aggregations/src/bin/aggregations_client.rs
@@ -12,23 +12,23 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let car_lifecycle = BasicValueDownlinkLifecycle::default()
         .on_event_blocking(|speed| println!("/cars/1 speed: {speed}"));
-    let _car_view = client_handle
+    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}"));
-    let _area_view = client_handle
+    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}"));
-    let _city_view = client_handle
+    client_handle
         .value_downlink::<f64>(RemotePath::new(host, "/city", "average_speed"))
         .lifecycle(city_lifecycle)
         .open()
diff --git a/example_apps/command_lane/Cargo.toml b/example_apps/command_lane/Cargo.toml
@@ -11,5 +11,5 @@ swimos_form = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[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
@@ -13,7 +13,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let state_lifecycle =
         BasicValueDownlinkLifecycle::default().on_event_blocking(|state| println!("{state}"));
-    let _state_view = client_handle
+    client_handle
         .value_downlink::<i32>(RemotePath::new(host, "/example/1", "lane"))
         .lifecycle(state_lifecycle)
         .open()
diff --git a/example_apps/demand_lane/Cargo.toml b/example_apps/demand_lane/Cargo.toml
@@ -10,5 +10,5 @@ swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[bin]]
 name = "demand_client"
diff --git a/example_apps/demand_lane/src/bin/demand_client.rs b/example_apps/demand_lane/src/bin/demand_client.rs
@@ -20,7 +20,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let demand_lifecycle =
         BasicValueDownlinkLifecycle::default().on_event_blocking(|state| println!("{state}"));
-    let _demand_view = client_handle
+    client_handle
         .value_downlink::<i32>(RemotePath::new(host, "/example/1", "demand"))
         .lifecycle(demand_lifecycle)
         .open()
diff --git a/example_apps/demand_map_lane/Cargo.toml b/example_apps/demand_map_lane/Cargo.toml
@@ -10,5 +10,5 @@ swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[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
@@ -17,7 +17,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let demand_lifecycle = BasicMapDownlinkLifecycle::default()
         .on_update_blocking(|key, _map, _old_state, new_state| println!("{key} -> {new_state}"));
-    let _demand_map_view = client_handle
+    client_handle
         .map_downlink::<String, i32>(RemotePath::new(host, "/example/1", "demand_map"))
         .lifecycle(demand_lifecycle)
         .open()
diff --git a/example_apps/http_lane/Cargo.toml b/example_apps/http_lane/Cargo.toml
@@ -8,9 +8,7 @@ publish = false
 swimos = { workspace = true, features = ["server", "agent"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
-
-[dev-dependencies]
 reqwest = { workspace = true }
 
-[[example]]
+[[bin]]
 name = "http_client"
diff --git a/example_apps/http_lane/src/bin/http_client.rs b/example_apps/http_lane/src/bin/http_client.rs
diff --git a/example_apps/join_map/Cargo.toml b/example_apps/join_map/Cargo.toml
@@ -10,5 +10,5 @@ swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[bin]]
 name = "join_map_client"
diff --git a/example_apps/join_map/src/agents.rs b/example_apps/join_map/src/agents.rs
@@ -34,8 +34,6 @@ impl AggregatedLifecycle {
         &self,
         context: HandlerContext<AggregatedStatisticsAgent>,
     ) -> impl EventHandler<AggregatedStatisticsAgent> {
-        println!("Start agg");
-
         let california_downlink = context.add_map_downlink(
             AggregatedStatisticsAgent::STREETS,
             "california".to_string(),
diff --git a/example_apps/join_map/src/bin/join_map_client.rs b/example_apps/join_map/src/bin/join_map_client.rs
@@ -13,7 +13,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let aggregated_lifecycle = BasicMapDownlinkLifecycle::default()
         .on_update_blocking(|key, _map, _old_state, new_state| println!("{key} -> {new_state}"));
-    let _aggregated_view = client_handle
+    client_handle
         .map_downlink::<String, u64>(RemotePath::new(host, "/join/state/example", "streets"))
         .lifecycle(aggregated_lifecycle)
         .open()
diff --git a/example_apps/join_value/Cargo.toml b/example_apps/join_value/Cargo.toml
@@ -9,9 +9,7 @@ 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 }
 
-[dev-dependencies]
-rand.workspace = true
-
-[[example]]
+[[bin]]
 name = "join_value_client"
diff --git a/example_apps/join_value/src/bin/join_value_client.rs b/example_apps/join_value/src/bin/join_value_client.rs
@@ -21,7 +21,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         let aggregated_lifecycle = BasicMapDownlinkLifecycle::default().on_update_blocking(
             |key, _map, _old_state, new_state| println!("Building {key} -> {new_state}"),
         );
-        let _aggregated_view = client_handle
+        client_handle
             .map_downlink::<String, bool>(RemotePath::new(
                 host,
                 format!("/buildings/{building}"),
diff --git a/example_apps/map_lane/Cargo.toml b/example_apps/map_lane/Cargo.toml
@@ -10,5 +10,5 @@ swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[bin]]
 name = "map_client"
diff --git a/example_apps/map_lane/src/bin/map_client.rs b/example_apps/map_lane/src/bin/map_client.rs
diff --git a/example_apps/supply_lane/Cargo.toml b/example_apps/supply_lane/Cargo.toml
@@ -10,5 +10,5 @@ swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[bin]]
 name = "supply_client"
diff --git a/example_apps/supply_lane/src/bin/supply_client.rs b/example_apps/supply_lane/src/bin/supply_client.rs
@@ -17,7 +17,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     let doubled_lifecycle = BasicEventDownlinkLifecycle::default().on_event_blocking(|event| {
         println!("{event:?}");
     });
-    let _doubled_view = client_handle
+    client_handle
         .event_downlink::<i32>(RemotePath::new(host, "/example/1", "supply"))
         .lifecycle(doubled_lifecycle)
         .open()
diff --git a/example_apps/value_lane/Cargo.toml b/example_apps/value_lane/Cargo.toml
@@ -10,5 +10,5 @@ swimos_client = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 example-util = { path = "../example_util" }
 
-[[example]]
+[[bin]]
 name = "value_client"
diff --git a/example_apps/value_lane/src/bin/value_client.rs b/example_apps/value_lane/src/bin/value_client.rs
diff --git a/runtime/swimos_runtime/src/downlink/mod.rs b/runtime/swimos_runtime/src/downlink/mod.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::HashSet;
+use std::error::Error;
 use std::fmt::Debug;
 use std::num::NonZeroUsize;
 use std::time::Duration;
@@ -511,10 +512,13 @@ impl<D: Decoder> DownlinkReceiver<D> {
     }
 }
 
-struct Failed<E>(u64, E);
+struct Failed(u64, Box<dyn Error + 'static>);
 
-impl<D: Decoder> Stream for DownlinkReceiver<D> {
-    type Item = Result<D::Item, Failed<D::Error>>;
+impl<D: Decoder> Stream for DownlinkReceiver<D>
+where
+    D::Error: Error + 'static,
+{
+    type Item = Result<D::Item, Failed>;
 
     fn poll_next(
         self: std::pin::Pin<&mut Self>,
@@ -526,7 +530,7 @@ impl<D: Decoder> Stream for DownlinkReceiver<D> {
         } else {
             this.receiver
                 .poll_next_unpin(cx)
-                .map_err(|e| Failed(this.id, e))
+                .map_err(|e| Failed(this.id, Box::new(e)))
         }
     }
 }
@@ -960,7 +964,7 @@ async fn write_task<B: DownlinkBackpressure>(
     mut backpressure: B,
     stop_voter: Voter,
 ) where
-    <<B as DownlinkBackpressure>::Dec as Decoder>::Error: Debug,
+    <<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() {
diff --git a/swimos_client/src/transport.rs b/swimos_client/src/transport.rs
@@ -32,7 +32,7 @@ use swimos_utilities::trigger;
 use tokio::select;
 use tokio::sync::{mpsc, oneshot};
 use tokio::task::JoinError;
-use tracing::{debug, error, trace};
+use tracing::{debug, error};
 
 type AttachCallback =
     oneshot::Sender<Result<(SocketAddr, mpsc::Sender<AttachClient>), DownlinkRuntimeError>>;
@@ -189,7 +189,6 @@ where
 
             match event {
                 TransportEvent::Request(TransportRequest::Resolve(shp, callback)) => {
-                    trace!(%shp, "Resolving host");
                     let shared_networking = &networking;
                     let resolve_fut = async move {
                         let result = shared_networking
@@ -206,17 +205,14 @@ where
                     addrs,
                     callback,
                 }) => {
-                    trace!(?host, ?scheme, "Resolving connection");
                     let peer = addrs
                         .iter()
                         .find_map(|sock| peers.get(sock).map(|tx| (sock, tx)));
                     match peer {
                         Some((sock, peer)) => {
-                            trace!("Found peer");
                             let _r = callback.send(Ok((*sock, peer.clone())));
                         }
                         None => {
-                            trace!("No peer found");
                             let shared_networking = &networking;
                             events.push(
                                 async move {
@@ -250,7 +246,6 @@ where
                     socket,
                     addr,
                 } => {
-                    trace!(?host, "Opening connection");
                     let shared_ws = &websockets;
                     let provider = &ext_provider;
                     let handshake_fut = async move {
@@ -281,7 +276,6 @@ where
                     callback,
                     websocket,
                 } => {
-                    trace!(?host, "Starting remote task");
                     let (attach_tx, attach_rx) = mpsc::channel(buffer_size.get());
                     let remote = RemoteTask::new(
                         remote_issuer.next_id(),
