Merge pull request #652 from swimos/downlink_stop

Adds stop functionality for downlinks in Join Map and Join Value lanes

Author: SirCipher

server/swimos_agent/src/agent_lifecycle/utility/mod.rs

@@ -20,12 +20,13 @@ use std::{collections::HashMap, marker::PhantomData};
 
 use futures::stream::unfold;
 use futures::{Future, FutureExt, Stream, StreamExt};
+use tokio::time::Instant;
+
 use swimos_api::address::Address;
 use swimos_form::read::RecognizerReadable;
 use swimos_form::write::StructuralWritable;
 use swimos_form::Form;
 use swimos_utilities::routing::RouteUri;
-use tokio::time::Instant;
 
 use crate::agent_model::downlink::hosted::{
     EventDownlinkHandle, MapDownlinkHandle, ValueDownlinkHandle,
@@ -43,7 +44,8 @@ use crate::event_handler::{
 };
 use crate::event_handler::{GetAgentUri, HandlerAction, SideEffect};
 use crate::item::{
-    MapLikeItem, MutableMapLikeItem, MutableValueLikeItem, TransformableMapLikeItem, ValueLikeItem,
+    JoinLikeItem, MapLikeItem, MutableMapLikeItem, MutableValueLikeItem, TransformableMapLikeItem,
+    ValueLikeItem,
 };
 use crate::lanes::command::{CommandLane, DoCommand};
 use crate::lanes::demand::{Cue, DemandLane};
@@ -649,7 +651,7 @@ impl<Agent: 'static> HandlerContext<Agent> {
     ///
     /// # Arguments
     /// * `lane` - Projection to the lane.
-    /// * `key - The key for the downlink.
+    /// * `key` - The key for the downlink.
     /// * `host` - The remote host at which the agent resides (a local agent if not specified).
     /// * `node` - The node URI of the agent.
     /// * `lane_uri` - The lane to downlink from.
@@ -670,12 +672,30 @@ impl<Agent: 'static> HandlerContext<Agent> {
         JoinValueAddDownlink::new(lane, key, address)
     }
 
+    /// Removes a downlink from a Join lane. Removing any associated values the downlink holds in
+    /// the underlying map.
+    ///
+    /// # Arguments
+    /// * `lane` - Projection to the lane.
+    /// * `link_key` - The link key for the downlink.
+    pub fn remove_downlink<L, K>(
+        &self,
+        lane: fn(&Agent) -> &L,
+        link_key: K,
+    ) -> impl HandlerAction<Agent, Completion = ()> + Send + 'static
+    where
+        K: Clone + Send + Eq + PartialEq + Hash + 'static,
+        L: JoinLikeItem<K>,
+    {
+        L::remove_downlink_handler(lane, link_key)
+    }
+
     /// Add a downlink to a Join Map lane. All key-value pairs received on the downlink will be set into the
     /// map state of the lane.
     ///
     /// # Arguments
     /// * `lane` - Projection to the lane.
-    /// * `link_key - A key to identify the link.
+    /// * `link_key` - A key to identify the link.
     /// * `host` - The remote host at which the agent resides (a local agent if not specified).
     /// * `node` - The node URI of the agent.
     /// * `lane_uri` - The lane to downlink from.

server/swimos_agent/src/agent_model/downlink/hosted/event/mod.rs

@@ -355,4 +355,9 @@ impl EventDownlinkHandle {
     pub fn is_linked(&self) -> bool {
         matches!(self.observer.get(), DlState::Linked | DlState::Synced)
     }
+
+    /// Consumes this handle and returns the stop handle, if it exists.
+    pub fn into_stop_rx(self) -> Option<trigger::Sender> {
+        self.stop_tx
+    }
 }

server/swimos_agent/src/item.rs

@@ -107,3 +107,22 @@ pub trait MutableValueLikeItem<T> {
 
     fn set_handler<C: 'static>(projection: fn(&C) -> &Self, value: T) -> Self::SetHandler<C>;
 }
+
+/// Trait for abstracting over common functionality between Join Map and Join Value lanes.
+pub trait JoinLikeItem<L> {
+    /// Handler action for removing a downlink from a join lane.
+    type RemoveDownlinkHandler<C>: HandlerAction<C, Completion = ()> + Send + 'static
+    where
+        C: 'static;
+
+    /// Create a handler that will remove a downlink from the lane and clear any entries in the
+    /// underlying map.
+    ///
+    /// # Arguments
+    /// * `projection`: a projection to the join lane.
+    /// * `link_key`: a key that signifies the downlink to remove.
+    fn remove_downlink_handler<C: 'static>(
+        projection: fn(&C) -> &Self,
+        link_key: L,
+    ) -> Self::RemoveDownlinkHandler<C>;
+}

server/swimos_agent/src/lanes/join/map/downlink/tests.rs

@@ -264,15 +264,19 @@ where
 fn state_for(
     lane: &JoinMapLane<String, i32, String>,
     link_key: &str,
-) -> (Option<Link<i32>>, HashSet<i32>) {
+) -> (Option<(DownlinkStatus, HashSet<i32>)>, HashSet<i32>) {
     let JoinMapLane { link_tracker, .. } = lane;
     let guard = link_tracker.borrow();
     let owned = guard
         .ownership
         .iter()
         .filter_map(|(k, v)| if v == link_key { Some(*k) } else { None })
         .collect();
-    (guard.links.get(link_key).cloned(), owned)
+    let state = guard
+        .links
+        .get(link_key)
+        .map(|link| (link.status, link.keys.clone()));
+    (state, owned)
 }
 
 fn set_state_for(
@@ -324,7 +328,7 @@ fn run_on_linked() {
     } else {
         panic!("Events incorrect: {:?}", events);
     }
-    if let (Some(Link { status, keys }), owned) = state_for(&agent.lane, "link") {
+    if let (Some((status, keys)), owned) = state_for(&agent.lane, "link") {
         assert_eq!(status, DownlinkStatus::Linked);
         assert!(keys.is_empty());
         assert!(owned.is_empty());
@@ -522,7 +526,7 @@ fn run_update_linked() {
     assert_eq!(value, Some("a".to_string()));
     let (state, owned) = state_for(&agent.lane, "link");
     let expected = [1].into_iter().collect();
-    if let Some(Link { status, keys }) = state {
+    if let Some((status, keys)) = state {
         assert_eq!(status, DownlinkStatus::Linked);
         assert_eq!(keys, expected);
     }
@@ -572,7 +576,7 @@ fn run_remove_linked() {
 
     let (state, owned) = state_for(&agent.lane, "link");
     let expected = [2].into_iter().collect();
-    if let Some(Link { status, keys }) = state {
+    if let Some((status, keys)) = state {
         assert_eq!(status, DownlinkStatus::Linked);
         assert_eq!(keys, expected);
     }
@@ -621,7 +625,7 @@ fn run_clear_linked() {
     assert!(value.is_none());
     let (state, owned) = state_for(&agent.lane, "link");
 
-    if let Some(Link { status, keys }) = state {
+    if let Some((status, keys)) = state {
         assert_eq!(status, DownlinkStatus::Linked);
         assert!(keys.is_empty());
     }
@@ -672,7 +676,7 @@ fn run_take_linked() {
     let expected = [1].into_iter().collect();
     let (state, owned) = state_for(&agent.lane, "link");
 
-    if let Some(Link { status, keys }) = state {
+    if let Some((status, keys)) = state {
         assert_eq!(status, DownlinkStatus::Linked);
         assert_eq!(keys, expected);
     }
@@ -723,7 +727,7 @@ fn run_drop_linked() {
     let expected = [2].into_iter().collect();
     let (state, owned) = state_for(&agent.lane, "link");
 
-    if let Some(Link { status, keys }) = state {
+    if let Some((status, keys)) = state {
         assert_eq!(status, DownlinkStatus::Linked);
         assert_eq!(keys, expected);
     }
@@ -782,7 +786,7 @@ fn run_on_synced() {
 
     let (state, owned) = state_for(&agent.lane, "link");
 
-    if let Some(Link { status, keys }) = state {
+    if let Some((status, keys)) = state {
         assert_eq!(status, DownlinkStatus::Linked);
         assert_eq!(keys, expected);
     }

server/swimos_agent/src/lanes/join/map/init/mod.rs

@@ -23,7 +23,6 @@ use crate::event_handler::{DowncastError, EventHandler, JoinLaneInitializer};
 use crate::lanes::JoinLaneKind;
 
 use super::AddDownlinkAction;
-//use super::AddDownlinkAction;
 use super::{lifecycle::JoinMapLaneLifecycle, JoinMapLane};
 
 /// Uses a [`JoinMapLaneLifecycle`] to create a handler action that will open a new downlink