Commander integration test.

Author: horned-sphere

server/swimos_agent/src/agent_model/mod.rs

@@ -30,9 +30,8 @@ use futures::{
 };
 use futures::{Future, FutureExt};
 use pin_project::pin_project;
-use swimos_agent_protocol::encoding::ad_hoc::CommandMessageEncoder;
 use swimos_agent_protocol::encoding::store::{RawMapStoreInitDecoder, RawValueStoreInitDecoder};
-use swimos_agent_protocol::{CommandMessage, LaneRequest, MapMessage};
+use swimos_agent_protocol::{LaneRequest, MapMessage};
 use swimos_api::agent::DownlinkKind;
 use swimos_api::agent::{HttpLaneRequest, LaneConfig, RawHttpLaneResponse};
 use swimos_api::error::{
@@ -52,7 +51,6 @@ use swimos_utilities::future::RetryStrategy;
 use swimos_utilities::routing::RouteUri;
 use tokio::io::AsyncWriteExt;
 use tokio::sync::mpsc;
-use tokio_util::codec::Encoder;
 use tracing::{debug, error, info, trace};
 use uuid::Uuid;
 
@@ -1064,9 +1062,8 @@ where
             mut suspended,
             link_requests:
                 LinkRequestCollector {
-                    next_commander_id,
                     downlinks: downlink_requests,
-                    commanders: commander_requests,
+                    commander_ids,
                 },
             mut join_lane_init,
             mut ad_hoc_buffer,
@@ -1078,7 +1075,7 @@ where
         let mut pending_writes = FuturesUnordered::new();
         let mut link_futures = FuturesUnordered::new();
         let mut external_item_ids_rev = HashMap::new();
-        let cmd_ids = RefCell::new(CommanderIds::new(next_commander_id));
+        let cmd_ids = RefCell::new(commander_ids);
 
         for (name, id) in external_item_ids.iter() {
             external_item_ids_rev.insert(*id, name);
@@ -1106,17 +1103,6 @@ where
             link_futures.push(LinkFuture::Opening(Some(request), fut));
         }
 
-        // Request commanders from the init phase.
-        let mut cmd_encoder = CommandMessageEncoder::default();
-        for request in commander_requests {
-            let CommanderRequest { path, id } = request;
-
-            let msg = CommandMessage::<_, ()>::register(path, id);
-            cmd_encoder
-                .encode(msg, &mut ad_hoc_buffer)
-                .expect("Encoding should be infallible.");
-        }
-
         for ((name, kind), (tx, rx)) in lane_io {
             if kind.map_like() {
                 let id = external_item_ids[&name];
@@ -1934,24 +1920,16 @@ async fn open_with_retry(
     }
 }
 
-#[derive(Debug)]
-struct CommanderRequest {
-    path: Address<Text>,
-    id: u16,
-}
-
 struct LinkRequestCollector<Context> {
-    next_commander_id: u16,
     downlinks: Vec<DownlinkSpawnRequest<Context>>,
-    commanders: Vec<CommanderRequest>,
+    commander_ids: CommanderIds,
 }
 
 impl<Context> Default for LinkRequestCollector<Context> {
     fn default() -> Self {
         Self {
-            next_commander_id: 0,
             downlinks: Default::default(),
-            commanders: Default::default(),
+            commander_ids: CommanderIds::default(),
         }
     }
 }
@@ -1960,7 +1938,7 @@ impl<Context> std::fmt::Debug for LinkRequestCollector<Context> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("LinkRequestCollector")
             .field("downlinks", &self.downlinks)
-            .field("commanders", &self.commanders)
+            .field("commander_ids", &self.commander_ids)
             .finish()
     }
 }
@@ -1982,14 +1960,7 @@ impl<Context> LinkSpawner<Context> for RefCell<LinkRequestCollector<Context>> {
 
     fn register_commander(&self, path: Address<Text>) -> Result<u16, CommanderRegistrationError> {
         let mut guard = self.borrow_mut();
-        let id = guard.next_commander_id;
-        if let Some(next) = guard.next_commander_id.checked_add(1) {
-            guard.next_commander_id = next;
-        } else {
-            return Err(CommanderRegistrationError::CommanderIdOverflow);
-        }
-        guard.commanders.push(CommanderRequest { path, id });
-        Ok(id)
+        guard.commander_ids.get_request(&path)
     }
 }
 
@@ -2017,19 +1988,13 @@ where
     }
 }
 
+#[derive(Default, Debug)]
 struct CommanderIds {
     next_id: u16,
     assigned: HashMap<Address<Text>, u16>,
 }
 
 impl CommanderIds {
-    fn new(next_id: u16) -> Self {
-        CommanderIds {
-            next_id,
-            assigned: HashMap::new(),
-        }
-    }
-
     fn get_request(&mut self, address: &Address<Text>) -> Result<u16, CommanderRegistrationError> {
         let id = if let Some(id) = self.assigned.get(address) {
             *id

server/swimos_agent/src/agent_model/tests/external_links/commander_lifecycle.rs

@@ -12,8 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use parking_lot::Mutex;
 use swimos_api::address::Address;
 use swimos_model::Text;
+use swimos_utilities::trigger;
 
 use crate::{
     agent_lifecycle::{
@@ -28,11 +32,21 @@ use super::empty_agent::EmptyAgent;
 
 pub struct CommanderLifecycle {
     address: Address<Text>,
+    stop_rx: Arc<Mutex<Option<trigger::Receiver>>>,
+}
+
+impl CommanderLifecycle {
+    pub fn new(address: Address<Text>, stop_rx: trigger::Receiver) -> Self {
+        CommanderLifecycle {
+            address,
+            stop_rx: Arc::new(Mutex::new(Some(stop_rx))),
+        }
+    }
 }
 
 impl CommanderLifecycle {
-    pub fn new(address: Address<Text>) -> Self {
-        CommanderLifecycle { address }
+    fn take_stop_rx(&self) -> trigger::Receiver {
+        self.stop_rx.lock().take().expect("Already taken.")
     }
 }
 
@@ -51,14 +65,23 @@ impl OnInit<EmptyAgent> for CommanderLifecycle {
 impl OnStart<EmptyAgent> for CommanderLifecycle {
     fn on_start(&self) -> impl EventHandler<EmptyAgent> + '_ {
         let context: HandlerContext<EmptyAgent> = Default::default();
+
+        let stop_rx = self.take_stop_rx();
+        let stop_handler = context.suspend(async move {
+            assert!(stop_rx.await.is_ok());
+            context.stop()
+        });
+
         let Address { host, node, lane } = &self.address;
         let create = context.create_commander(host.as_ref(), node, lane);
 
-        create.and_then(move |commander: Commander<EmptyAgent>| {
+        let commands = create.and_then(move |commander: Commander<EmptyAgent>| {
             commander
                 .send(7)
                 .followed_by(context.suspend(async move { commander.send(22) }))
-        })
+        });
+
+        stop_handler.followed_by(commands)
     }
 }