Merge pull request #624 from swimos/supply_lane_runtime

Fixes incorrect dispatching of event envelopes during downlink synchronisation

Author: SirCipher

client/swimos_client/src/lib.rs

@@ -15,6 +15,7 @@
 #[cfg(not(feature = "deflate"))]
 use ratchet::NoExtProvider;
 use ratchet::WebSocketStream;
+use std::marker::PhantomData;
 use std::num::NonZeroUsize;
 use swimos_remote::ws::RatchetClient;
 
@@ -28,12 +29,12 @@ pub use runtime::{CommandError, Commander, RemotePath};
 use std::sync::Arc;
 pub use swimos_api::downlink::DownlinkConfig;
 pub use swimos_downlink::lifecycle::{
-    BasicMapDownlinkLifecycle, BasicValueDownlinkLifecycle, MapDownlinkLifecycle,
-    ValueDownlinkLifecycle,
+    BasicEventDownlinkLifecycle, BasicMapDownlinkLifecycle, BasicValueDownlinkLifecycle,
+    EventDownlinkLifecycle, MapDownlinkLifecycle, ValueDownlinkLifecycle,
 };
 use swimos_downlink::{
-    ChannelError, DownlinkTask, MapDownlinkHandle, MapDownlinkModel, MapKey, MapValue,
-    NotYetSyncedError, ValueDownlinkModel, ValueDownlinkSet,
+    ChannelError, DownlinkTask, EventDownlinkModel, MapDownlinkHandle, MapDownlinkModel, MapKey,
+    MapValue, NotYetSyncedError, ValueDownlinkModel, ValueDownlinkSet,
 };
 use swimos_form::Form;
 use swimos_remote::net::dns::Resolver;
@@ -235,6 +236,24 @@ impl ClientHandle {
         }
     }
 
+    /// Returns an event downlink builder initialised with the default options.
+    ///
+    /// # Arguments
+    /// * `path` - The path of the downlink top open.
+    pub fn event_downlink<T>(
+        &self,
+        path: RemotePath,
+    ) -> EventDownlinkBuilder<'_, BasicEventDownlinkLifecycle<T>> {
+        EventDownlinkBuilder {
+            handle: self,
+            lifecycle: BasicEventDownlinkLifecycle::default(),
+            path,
+            options: DownlinkOptions::SYNC,
+            runtime_config: Default::default(),
+            downlink_config: Default::default(),
+        }
+    }
+
     /// Returns a map downlink builder initialised with the default options.
     ///
     /// # Arguments
@@ -497,3 +516,94 @@ impl<K, V> MapDownlinkView<K, V> {
         self.stop_rx.clone()
     }
 }
+
+/// A builder for value downlinks.
+pub struct EventDownlinkBuilder<'h, L> {
+    handle: &'h ClientHandle,
+    lifecycle: L,
+    path: RemotePath,
+    options: DownlinkOptions,
+    runtime_config: DownlinkRuntimeConfig,
+    downlink_config: DownlinkConfig,
+}
+
+impl<'h, L> EventDownlinkBuilder<'h, L> {
+    /// Sets a new lifecycle that to be used.
+    pub fn lifecycle<NL>(self, lifecycle: NL) -> EventDownlinkBuilder<'h, NL> {
+        let EventDownlinkBuilder {
+            handle,
+            path,
+            options,
+            runtime_config,
+            downlink_config,
+            ..
+        } = self;
+        EventDownlinkBuilder {
+            handle,
+            lifecycle,
+            path,
+            options,
+            runtime_config,
+            downlink_config,
+        }
+    }
+
+    /// Sets link options for the downlink.
+    pub fn options(mut self, options: DownlinkOptions) -> Self {
+        self.options = options;
+        self
+    }
+
+    /// Sets a new downlink runtime configuration.
+    pub fn runtime_config(mut self, config: DownlinkRuntimeConfig) -> Self {
+        self.runtime_config = config;
+        self
+    }
+
+    /// Sets a new downlink configuration.
+    pub fn downlink_config(mut self, config: DownlinkConfig) -> Self {
+        self.downlink_config = config;
+        self
+    }
+
+    /// Attempts to open the downlink.
+    pub async fn open<T>(self) -> Result<EventDownlinkView<T>, Arc<DownlinkRuntimeError>>
+    where
+        L: EventDownlinkLifecycle<T> + Sync + 'static,
+        T: Send + Sync + Form + Clone + 'static,
+        T::Rec: Send,
+    {
+        let EventDownlinkBuilder {
+            handle,
+            lifecycle,
+            path,
+            options,
+            runtime_config,
+            downlink_config,
+        } = self;
+        let task = DownlinkTask::new(EventDownlinkModel::new(lifecycle));
+        let stop_rx = handle
+            .inner
+            .run_downlink(path, runtime_config, downlink_config, options, task)
+            .await?;
+
+        Ok(EventDownlinkView {
+            _type: Default::default(),
+            stop_rx,
+        })
+    }
+}
+
+/// An event downlink handle which provides the functionality to await the downlink terminating.
+#[derive(Debug, Clone)]
+pub struct EventDownlinkView<T> {
+    _type: PhantomData<T>,
+    stop_rx: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+}
+
+impl<T> EventDownlinkView<T> {
+    /// Returns a receiver that completes with the result of downlink's internal task.
+    pub fn stop_notification(&self) -> promise::Receiver<Result<(), DownlinkRuntimeError>> {
+        self.stop_rx.clone()
+    }
+}

runtime/swimos_runtime/src/downlink/mod.rs

@@ -555,6 +555,10 @@ where
     let mut awaiting_synced: Vec<DownlinkSender> = vec![];
     let mut awaiting_linked: Vec<DownlinkSender> = vec![];
     let mut registered: Vec<DownlinkSender> = vec![];
+    // Track whether any events have been received while syncing the downlink. While this isn't the
+    // nicest thing to have, it's required to distinguish between communicating with a stateless
+    // lane and a downlink syncing with a lane that has a type which may be optional.
+    let mut sync_event = false;
 
     let result: Result<(), H::Report> = loop {
         let (event, is_active) = match task_state.as_mut().as_pin_mut() {
@@ -653,7 +657,17 @@ where
                     trace!("Entering Synced state.");
                     dl_state = ReadTaskDlState::Synced;
                     if is_active {
-                        if I::SINGLE_FRAME_STATE {
+                        // `sync_event` will be false if we're communicating with a stateless lane
+                        // as no event envelope will have been sent. However, it's valid Recon for
+                        // an empty event envelope to be sent (consider Option::None) and this must
+                        // still be sent to the downlink task.
+                        //
+                        // If we're linked to a stateless lane, then `sync_current` cannot be used
+                        // as we will not have received an event envelope as it will dispatch one
+                        // with the empty buffer and this may cause the downlink task's decoder to
+                        // fail due to reading an extant read event. Therefore, delegate the operation to
+                        // `sync_only` which will not send an event notification.
+                        if I::SINGLE_FRAME_STATE && sync_event {
                             sync_current(&mut awaiting_synced, &mut registered, &current).await;
                         } else {
                             sync_only(&mut awaiting_synced, &mut registered).await;
@@ -669,8 +683,11 @@ where
                     break Ok(());
                 }
                 Notification::Event(bytes) => {
+                    sync_event = true;
+
                     trace!("Updating the current value.");
                     current.clear();
+
                     if let Err(e) = interpretation.interpret_frame_data(bytes, &mut current) {
                         if let BadFrameResponse::Abort(report) = failure_handler.failed_with(e) {
                             break Err(report);

runtime/swimos_runtime/src/downlink/tests/value.rs

@@ -384,7 +384,7 @@ async fn shutdowm_after_attached() {
 }
 
 #[tokio::test]
-async fn shutdowm_after_corrupted_frame() {
+async fn shutdown_after_corrupted_frame() {
     let (events, result) = run_test(DownlinkOptions::empty(), |context| async move {
         let TestContext {
             mut tx,
@@ -731,7 +731,7 @@ async fn exhaust_output_buffer() {
 }
 
 #[tokio::test]
-async fn shutdowm_after_timeout_with_no_subscribers() {
+async fn shutdown_after_timeout_with_no_subscribers() {
     let ((_stop, events), result) = run_test_with_config(
         DownlinkOptions::empty(),
         DownlinkRuntimeConfig {
@@ -1079,3 +1079,100 @@ async fn receive_from_two_consumers() {
     assert_eq!(first_events, vec![DownlinkNotification::Unlinked]);
     assert_eq!(second_events, vec![DownlinkNotification::Unlinked]);
 }
+
+#[tokio::test]
+async fn sync_no_event() {
+    let (events, result) = run_test(
+        DownlinkOptions::SYNC,
+        |TestContext {
+             mut tx,
+             mut rx,
+             start_client,
+             stop,
+             mut events,
+             ..
+         }| async move {
+            expect_message(rx.recv().await, Operation::Link);
+
+            tx.link().await;
+
+            start_client.trigger();
+
+            expect_event(
+                events.next().await,
+                State::Unlinked,
+                DownlinkNotification::Linked,
+            );
+            expect_message(rx.recv().await, Operation::Sync);
+
+            tx.sync().await;
+
+            expect_event(
+                events.next().await,
+                State::Linked,
+                DownlinkNotification::Synced,
+            );
+
+            stop.trigger();
+            events.collect::<Vec<_>>().await
+        },
+    )
+    .await;
+    assert!(result.is_ok());
+    assert_eq!(
+        events,
+        vec![(State::Synced, DownlinkNotification::Unlinked)]
+    );
+}
+
+#[tokio::test]
+async fn sync_empty_event() {
+    let (events, result) = run_test(
+        DownlinkOptions::SYNC,
+        |TestContext {
+             mut tx,
+             mut rx,
+             start_client,
+             stop,
+             mut events,
+             ..
+         }| async move {
+            expect_message(rx.recv().await, Operation::Link);
+
+            tx.link().await;
+
+            start_client.trigger();
+
+            expect_event(
+                events.next().await,
+                State::Unlinked,
+                DownlinkNotification::Linked,
+            );
+            expect_message(rx.recv().await, Operation::Sync);
+
+            let content = Message::CurrentValue(Text::new(""));
+            tx.update(content.clone()).await;
+            tx.sync().await;
+
+            expect_event(
+                events.next().await,
+                State::Linked,
+                DownlinkNotification::Event { body: content },
+            );
+            expect_event(
+                events.next().await,
+                State::Linked,
+                DownlinkNotification::Synced,
+            );
+
+            stop.trigger();
+            events.collect::<Vec<_>>().await
+        },
+    )
+    .await;
+    assert!(result.is_ok());
+    assert_eq!(
+        events,
+        vec![(State::Synced, DownlinkNotification::Unlinked)]
+    );
+}