Merge pull request #617 from swimos/map_downlink_fix

Resolves incorrect map downlink codec usage.

Author: SirCipher

client/runtime/src/tests.rs

@@ -581,7 +581,7 @@ where
     let downlink = TrackingMapDownlink::new(
         spawned.clone(),
         stopped.clone(),
-        MapDownlinkModel::new(set_rx, lifecycle, false),
+        MapDownlinkModel::new(set_rx, lifecycle),
     );
 
     let promise = handle

client/swimos_client/src/lib.rs

@@ -446,7 +446,7 @@ impl<'h, L> MapDownlinkBuilder<'h, L> {
         } = self;
 
         let (tx, rx) = mpsc::channel(downlink_config.buffer_size.get());
-        let task = DownlinkTask::new(MapDownlinkModel::new(rx, lifecycle, true));
+        let task = DownlinkTask::new(MapDownlinkModel::new(rx, lifecycle));
         let stop_rx = handle
             .inner
             .run_downlink(path, runtime_config, downlink_config, options, task)

swimos_downlink/src/model/mod.rs

@@ -64,20 +64,14 @@ impl<T, LC> EventDownlinkModel<T, LC> {
 pub struct MapDownlinkModel<K, V, LC> {
     pub actions: mpsc::Receiver<MapMessage<K, V>>,
     pub lifecycle: LC,
-    pub remote: bool,
 }
 
 impl<K, V, LC> MapDownlinkModel<K, V, LC> {
     pub fn new(
         actions: mpsc::Receiver<MapMessage<K, V>>,
         lifecycle: LC,
-        remote: bool,
     ) -> MapDownlinkModel<K, V, LC> {
-        MapDownlinkModel {
-            actions,
-            lifecycle,
-            remote,
-        }
+        MapDownlinkModel { actions, lifecycle }
     }
 }
 
@@ -105,9 +99,8 @@ pub fn event_downlink<T>() -> DefaultEventDownlinkModel<T> {
 
 pub fn map_downlink<K, V>(
     actions: mpsc::Receiver<MapMessage<K, V>>,
-    remote: bool,
 ) -> DefaultMapDownlinkModel<K, V> {
-    MapDownlinkModel::new(actions, Default::default(), remote)
+    MapDownlinkModel::new(actions, Default::default())
 }
 
 #[derive(Debug)]
@@ -220,16 +213,11 @@ where
         F: Fn(LC) -> LC2,
         LC2: MapDownlinkLifecycle<K, V>,
     {
-        let MapDownlinkModel {
-            actions,
-            lifecycle,
-            remote,
-        } = self;
+        let MapDownlinkModel { actions, lifecycle } = self;
 
         MapDownlinkModel {
             actions,
             lifecycle: f(lifecycle),
-            remote,
         }
     }
 }

swimos_downlink/src/task/map.rs

@@ -23,7 +23,6 @@ use swimos_api::downlink::DownlinkConfig;
 use swimos_api::error::DownlinkTaskError;
 use swimos_api::protocol::downlink::{
     DownlinkNotification, DownlinkOperation, DownlinkOperationEncoder, MapNotificationDecoder,
-    ValueNotificationDecoder,
 };
 use swimos_api::protocol::map::MapMessage;
 use swimos_form::structural::write::StructuralWritable;
@@ -36,7 +35,7 @@ use tokio::select;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 use tokio_util::codec::{Decoder, FramedRead, FramedWrite};
-use tracing::{info_span, trace, Instrument};
+use tracing::{error, info_span, trace, Instrument};
 
 /// Task to drive a map downlink, calling lifecycle events at appropriate points.
 ///
@@ -60,35 +59,18 @@ where
     V::Rec: Send,
     LC: MapDownlinkLifecycle<K, V>,
 {
-    let MapDownlinkModel {
-        actions,
-        lifecycle,
-        remote,
-    } = model;
+    let MapDownlinkModel { actions, lifecycle } = model;
 
-    if remote {
-        run_io(
-            config,
-            input,
-            lifecycle,
-            FramedWrite::new(output, DownlinkOperationEncoder),
-            actions,
-            ValueNotificationDecoder::default(),
-        )
-        .instrument(info_span!("Downlink IO task.", %path))
-        .await
-    } else {
-        run_io(
-            config,
-            input,
-            lifecycle,
-            FramedWrite::new(output, DownlinkOperationEncoder),
-            actions,
-            MapNotificationDecoder::default(),
-        )
-        .instrument(info_span!("Downlink IO task.", %path))
-        .await
-    }
+    run_io(
+        config,
+        input,
+        lifecycle,
+        FramedWrite::new(output, DownlinkOperationEncoder),
+        actions,
+        MapNotificationDecoder::default(),
+    )
+    .instrument(info_span!("Downlink IO task.", %path))
+    .await
 }
 
 /// The current state of the downlink.
@@ -141,6 +123,7 @@ where
     V::Rec: Send,
     LC: MapDownlinkLifecycle<K, V>,
     Snk: Sink<DownlinkOperation<MapMessage<K, V>>> + Unpin,
+    Snk::Error: Debug,
     D: Decoder<Item = DownlinkNotification<MapMessage<K, V>>, Error = E>,
     DownlinkTaskError: From<E>,
     E: Debug,
@@ -159,9 +142,7 @@ where
                     write = (&mut write_fut) => IoEvent::Write(write),
                     read_event = framed_read.next() => match read_event {
                         Some(Ok(notification)) => IoEvent::Read(notification),
-                        Some(Err(e)) => {
-                            break Err(e.into())
-                        } ,
+                        Some(Err(e)) => break Err(e.into()),
                         None => break Ok(()),
                     }
                 };
@@ -196,8 +177,8 @@ where
                         }
 
                         let op = DownlinkOperation::new(message);
-                        if framed.feed(op).await.is_err() {
-                            mode = Mode::Read;
+                        if let Err(e) = framed.feed(op).await {
+                            error!(error = ?e, "Failed to feed downlink frame. Transitioning to read-only mode");
                         }
                     }
                     IoEvent::Write(None) => mode = Mode::Read,

swimos_downlink/src/task/tests/event.rs

@@ -20,7 +20,7 @@ use swimos_api::{downlink::DownlinkConfig, protocol::downlink::DownlinkNotificat
 use swimos_utilities::non_zero_usize;
 use tokio::sync::mpsc;
 
-use super::run_downlink_task;
+use super::run_value_downlink_task;
 use crate::model::lifecycle::{BasicEventDownlinkLifecycle, EventDownlinkLifecycle};
 use crate::{DownlinkTask, EventDownlinkModel};
 
@@ -69,7 +69,7 @@ async fn link_downlink() {
         buffer_size: DEEFAULT_BUFFER_SIZE,
     };
 
-    let result = run_downlink_task(
+    let result = run_value_downlink_task(
         DownlinkTask::new(model),
         config,
         |mut writer, reader| async move {
@@ -96,7 +96,7 @@ async fn message_before_linked() {
         buffer_size: DEEFAULT_BUFFER_SIZE,
     };
 
-    let result = run_downlink_task(
+    let result = run_value_downlink_task(
         DownlinkTask::new(model),
         config,
         |mut writer, reader| async move {
@@ -126,7 +126,7 @@ async fn message_after_linked() {
         buffer_size: DEEFAULT_BUFFER_SIZE,
     };
 
-    let result = run_downlink_task(
+    let result = run_value_downlink_task(
         DownlinkTask::new(model),
         config,
         |mut writer, reader| async move {
@@ -157,7 +157,7 @@ async fn terminate_after_unlinked() {
         buffer_size: DEEFAULT_BUFFER_SIZE,
     };
 
-    let result = run_downlink_task(
+    let result = run_value_downlink_task(
         DownlinkTask::new(model),
         config,
         |mut writer, reader| async move {
@@ -197,7 +197,7 @@ async fn terminate_after_corrupt_frame() {
         buffer_size: DEEFAULT_BUFFER_SIZE,
     };
 
-    let result = run_downlink_task(
+    let result = run_value_downlink_task(
         DownlinkTask::new(model),
         config,
         |mut writer, reader| async move {
@@ -228,7 +228,7 @@ async fn relink_downlink() {
         buffer_size: DEEFAULT_BUFFER_SIZE,
     };
 
-    let result = run_downlink_task(
+    let result = run_value_downlink_task(
         DownlinkTask::new(model),
         config,
         |mut writer, reader| async move {