Merge branch 'main' of https://github.com/swimos/swim-rust into fluvio

Author: SirCipher

server/swimos_connector/src/connector/egress/mod.rs

@@ -142,6 +142,11 @@ pub trait EgressConnectorSender<SendError>: Send + Clone {
 /// A reference to an egress context is passed to an [egress connector](`EgressConnector`) when it starts
 /// allowing it to request that lanes or downlinks to remote lanes be opened.
 pub trait EgressContext {
+    /// Request a new, dynamic WARP lane be opened on the agent.
+    ///
+    /// # Arguments
+    /// * `name` - The name of the lane.
+    /// * `kind` - The kind of the lane.
     fn open_lane(&mut self, name: &str, kind: WarpLaneKind);
 
     /// Request an event downlink to a remote lane.

server/swimos_connector/src/connector/ingress/mod.rs

@@ -98,5 +98,10 @@ where
 /// A reference to an ingress context is passed to an [ingress connector](`IngressConnector`) when it starts
 /// allowing it to request that lanes be opened.
 pub trait IngressContext {
+    /// Request a new, dynamic WARP lane be opened on the agent.
+    ///
+    /// # Arguments
+    /// * `name` - The name of the lane.
+    /// * `kind` - The kind of the lane.
     fn open_lane(&mut self, name: &str, kind: WarpLaneKind);
 }

server/swimos_connector_kafka/src/config/egress.rs

@@ -95,9 +95,9 @@ impl From<&DownlinkAddress> for Address<String> {
     }
 }
 
-impl DownlinkAddress {
-    pub fn borrow_as_addr(&self) -> Address<&str> {
-        let DownlinkAddress { host, node, lane } = self;
+impl<'a> From<&'a DownlinkAddress> for Address<&'a str> {
+    fn from(value: &'a DownlinkAddress) -> Self {
+        let DownlinkAddress { host, node, lane } = value;
         Address {
             host: host.as_ref().map(|s| s.as_str()),
             node,

server/swimos_connector_kafka/src/connector/egress/mod.rs

@@ -251,10 +251,10 @@ fn open_downlinks(config: &KafkaEgressConfiguration, context: &mut dyn EgressCon
         ..
     } = config;
     for value_dl in value_downlinks {
-        context.open_event_downlink(value_dl.address.borrow_as_addr());
+        context.open_event_downlink(<Address<&str>>::from(&value_dl.address));
     }
     for map_dl in map_downlinks {
-        context.open_map_downlink(map_dl.address.borrow_as_addr());
+        context.open_map_downlink(<Address<&str>>::from(&map_dl.address));
     }
 }
 

server/swimos_connector_kafka/src/connector/egress/tests/integration.rs

@@ -277,8 +277,8 @@ fn initialize_connector() {
     .into_iter()
     .collect::<HashMap<_, _>>();
 
-    assert_eq!(value_downlinks, vec![Address::from(&addr1())]);
-    assert_eq!(map_downlinks, vec![Address::from(&addr2())]);
+    assert_eq!(value_downlinks, vec![Address::<String>::from(&addr1())]);
+    assert_eq!(map_downlinks, vec![Address::<String>::from(&addr2())]);
     assert_eq!(lanes_map, expected_lanes);
 }
 

server/swimos_connector_kafka/src/connector/egress/tests/mod.rs

@@ -87,32 +87,6 @@ fn downlinks_config() -> KafkaEgressConfiguration {
     }
 }
 
-/* #[tokio::test]
-async fn open_lanes() {
-    let config = lanes_config();
-    let agent = ConnectorAgent::default();
-    let selectors = MessageSelectors::try_from(&config).expect("Bad configuration.");
-
-    let semaphore = Arc::new(Semaphore::new(0));
-    let (done_tx, done_rx) = trigger::trigger();
-    let handler = selectors.open_lanes(done_tx, semaphore, 0);
-
-    let handler_task = run_handler_with_futures(&agent, handler);
-
-    let (modified, done_result) = timeout(TEST_TIMEOUT, join(handler_task, done_rx))
-        .await
-        .expect("Test timed out.");
-
-    assert!(modified.is_empty());
-    assert!(done_result.is_ok());
-
-    let expected_value_lanes = [VALUE_LANE.to_string()].into_iter().collect::<HashSet<_>>();
-    let expected_map_lanes = [MAP_LANE.to_string()].into_iter().collect::<HashSet<_>>();
-
-    assert_eq!(agent.value_lanes(), expected_value_lanes);
-    assert_eq!(agent.map_lanes(), expected_map_lanes);
-} */
-
 #[derive(Default, Debug, PartialEq, Eq)]
 struct TestEgressContext {
     lanes: Vec<(String, WarpLaneKind)>,