Added end to end test for MQTT ingress connector.

Author: horned-sphere

server/swimos_connector/src/config/ingress.rs

@@ -91,6 +91,17 @@ pub struct ValueRelaySpecification {
     pub required: bool,
 }
 
+impl ValueRelaySpecification {
+    pub fn new<S: Into<String>>(node: S, lane: S, payload: S, required: bool) -> Self {
+        ValueRelaySpecification {
+            node: node.into(),
+            lane: lane.into(),
+            payload: payload.into(),
+            required,
+        }
+    }
+}
+
 /// Specification of a map relay for the connector.
 #[derive(Clone, Debug, Form, PartialEq, Eq)]
 #[form(tag = "MapRelaySpec")]
@@ -111,6 +122,26 @@ pub struct MapRelaySpecification {
     pub remove_when_no_value: bool,
 }
 
+impl MapRelaySpecification {
+    pub fn new<S: Into<String>>(
+        node: S,
+        lane: S,
+        key: S,
+        value: S,
+        required: bool,
+        remove_when_no_value: bool,
+    ) -> Self {
+        MapRelaySpecification {
+            node: node.into(),
+            lane: lane.into(),
+            key: key.into(),
+            value: value.into(),
+            required,
+            remove_when_no_value,
+        }
+    }
+}
+
 /// Specification of a relay for the connector.
 #[derive(Clone, Debug, Form, PartialEq, Eq)]
 pub enum RelaySpecification {

server/swimos_connector_mqqt/src/connector/ingress/tests/end_to_end.rs

@@ -0,0 +1,172 @@
+// Copyright 2015-2024 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::time::Duration;
+
+use bytes::BytesMut;
+use futures::{future::join, TryStreamExt};
+use swimos_agent::agent_model::{AgentSpec, ItemDescriptor, ItemFlags};
+use swimos_agent_protocol::{encoding::command::CommandMessageDecoder, CommandMessage};
+use swimos_api::{address::Address, agent::WarpLaneKind};
+use swimos_connector::{
+    config::{
+        format::DataFormat, IngressValueLaneSpec, RelaySpecification, ValueRelaySpecification,
+    },
+    BaseConnector, ConnectorAgent, IngressConnector, IngressContext,
+};
+use swimos_connector_util::{run_handler_with_futures, run_handler_with_futures_and_cmds};
+use swimos_model::Value;
+use swimos_utilities::trigger;
+use tokio::time::timeout;
+use tokio_util::codec::Decoder;
+
+use crate::{
+    facade::MqttFactory,
+    generator::{generate_data, TOPIC},
+    MqttIngressConfiguration, MqttIngressConnector, Subscription,
+};
+
+const CLIENT_URL: &str = "mqtt://localhost:1883?client_id=test";
+const GEN_URL: &str = "mqtt://localhost:1883?client_id=generator";
+const LANE_NAME: &str = "value_lane";
+
+fn make_configuration() -> MqttIngressConfiguration {
+    MqttIngressConfiguration {
+        url: CLIENT_URL.to_string(),
+        value_lanes: vec![IngressValueLaneSpec::new(Some(LANE_NAME), "$payload", true)],
+        map_lanes: vec![],
+        relays: vec![RelaySpecification::Value(ValueRelaySpecification::new(
+            "/node", "lane", "$payload", true,
+        ))],
+        payload_deserializer: DataFormat::String,
+        subscription: Subscription::Topic(TOPIC.to_string()),
+        keep_alive_secs: None,
+        max_packet_size: None,
+        channel_size: Some(0),
+        credentials: None,
+    }
+}
+
+struct TestContext<'a> {
+    agent: &'a ConnectorAgent,
+}
+
+impl<'a> IngressContext for TestContext<'a> {
+    fn open_lane(&mut self, name: &str, kind: WarpLaneKind) {
+        let TestContext { agent } = self;
+        agent
+            .register_dynamic_item(
+                name,
+                ItemDescriptor::WarpLane {
+                    kind,
+                    flags: ItemFlags::TRANSIENT,
+                },
+            )
+            .expect("Registering lane failed.");
+    }
+}
+
+async fn init_connector(agent: &ConnectorAgent) -> MqttIngressConnector<MqttFactory> {
+    let config = make_configuration();
+    let connector = MqttIngressConnector::for_config(config);
+
+    let mut init_context = TestContext { agent };
+    connector
+        .initialize(&mut init_context)
+        .expect("Initialization failed.");
+
+    let (done_tx, done_rx) = trigger::trigger();
+    let handler = connector.on_start(done_tx);
+
+    assert!(run_handler_with_futures(agent, handler).await.is_empty());
+    done_rx.await.expect("Starting connector failed.");
+    connector
+}
+
+async fn drive_connector(
+    agent: &mut ConnectorAgent,
+    connector: &MqttIngressConnector<MqttFactory>,
+    events: usize,
+) {
+    let mut stream = connector.create_stream().expect("Creating stream failed.");
+    let mut command_buffer = BytesMut::new();
+    let mut prev = None;
+    for _ in 0..events {
+        let handler = stream
+            .try_next()
+            .await
+            .expect("Receiving message failed.")
+            .expect("Stream stopped.");
+        command_buffer.clear();
+        run_handler_with_futures_and_cmds(agent, handler, &mut command_buffer).await;
+        check_value(&mut prev, agent, &mut command_buffer);
+    }
+}
+
+fn check_value(
+    prev: &mut Option<String>,
+    agent: &mut ConnectorAgent,
+    command_buffer: &mut BytesMut,
+) {
+    let guard = agent.value_lane(LANE_NAME).expect("Lane not defined");
+    let mut decoder = CommandMessageDecoder::<String, Value>::default();
+    let expected_addr = Address::new(None, "/node", "lane").owned();
+    match guard.read(Value::clone) {
+        Value::Text(s) => {
+            let string = s.to_string();
+            if let Some(p) = prev.take() {
+                assert_ne!(string, p);
+            }
+            let cmd_message = decoder
+                .decode_eof(command_buffer)
+                .expect("Decoding failed.")
+                .expect("Expected command.");
+            match cmd_message {
+                CommandMessage::Addressed {
+                    target,
+                    command,
+                    overwrite_permitted,
+                } => {
+                    assert_eq!(target, expected_addr);
+                    assert_eq!(command, Value::text(&string));
+                    assert!(!overwrite_permitted);
+                }
+                ow => panic!("Unexpected command message: {:?}", ow),
+            }
+            *prev = Some(string);
+        }
+        _ => panic!("Expected text."),
+    }
+}
+
+async fn connector_task(done_tx: trigger::Sender) {
+    let mut agent = ConnectorAgent::default();
+    let connector = init_connector(&agent).await;
+    drive_connector(&mut agent, &connector, 10).await;
+    done_tx.trigger();
+}
+
+const TEST_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[tokio::test]
+#[ignore] // Ignored by default as this relies on an external service being present.
+async fn run_ingress_connector() {
+    let (done_tx, done_rx) = trigger::trigger();
+    let run_connector = connector_task(done_tx);
+    let generator = generate_data(GEN_URL.to_string(), done_rx);
+    let (gen_result, _) = timeout(TEST_TIMEOUT, join(generator, run_connector))
+        .await
+        .expect("Timed out.");
+    assert!(gen_result.is_ok());
+}

server/swimos_connector_mqqt/src/connector/ingress/tests/mod.rs

@@ -12,5 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod end_to_end;
 mod integration;
 mod mock;

server/swimos_connector_mqqt/src/generator.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::pin::pin;
+use std::time::Duration;
 
 use bytes::Bytes;
 use futures::future::join;
@@ -21,24 +21,30 @@ use rumqttc::{ClientError, MqttOptions, QoS};
 use swimos_utilities::trigger;
 use tracing::debug;
 
-async fn generate_data(mqtt_uri: String, mut stop: trigger::Receiver) -> Result<(), Box<dyn std::error::Error>> {
+pub const TOPIC: &str = "test/topic";
 
+pub async fn generate_data(
+    mqtt_uri: String,
+    mut stop: trigger::Receiver,
+) -> Result<(), Box<dyn std::error::Error>> {
     let opts = MqttOptions::parse_url(mqtt_uri)?;
 
     let (client, mut event_loop) = rumqttc::AsyncClient::new(opts, 0);
     let stop_cpy = stop.clone();
-    let events_task = async move {    
+    let events_task = async move {
         loop {
-           tokio::select! {
-                biased;
-                _ = &mut stop => break Ok(()),
-                event = event_loop.poll() => {
-                    match event {
-                        Ok(ev) => debug!(event = ?ev, "Processed MQTT event."),
-                        Err(err) => break Err(err),
-                    }
-                }
-           }
+            tokio::select! {
+                 biased;
+                 _ = &mut stop => break Ok(()),
+                 event = event_loop.poll() => {
+                     match event {
+                         Ok(ev) => debug!(event = ?ev, "Processed MQTT event."),
+                         Err(err) => {
+                             break Err(err);
+                         }
+                     }
+                 }
+            }
         }
     };
 
@@ -52,7 +58,10 @@ async fn generate_data(mqtt_uri: String, mut stop: trigger::Receiver) -> Result<
             for _ in 0..10 {
                 data.push(rand.gen_range('A'..='Z'));
             }
-            client.publish_bytes("test/topic", QoS::AtMostOnce, true, Bytes::from(data)).await?;
+            client
+                .publish_bytes(TOPIC, QoS::AtMostOnce, true, Bytes::from(data))
+                .await?;
+            tokio::time::sleep(Duration::from_millis(100)).await;
         }
         Ok(()) as Result<(), ClientError>
     };
@@ -61,4 +70,4 @@ async fn generate_data(mqtt_uri: String, mut stop: trigger::Receiver) -> Result<
     ev_res?;
     gen_res?;
     Ok(())
-}
+}

server/swimos_connector_mqqt/src/lib.rs

@@ -25,4 +25,4 @@ pub use connector::{MqttEgressConnector, MqttIngressConnector, MqttSender};
 pub use error::MqttConnectorError;
 
 #[cfg(test)]
-mod generator;
+mod generator;

server/swimos_connector_util/src/lib.rs

@@ -164,26 +164,47 @@ where
     A: AgentSpec,
 {
     let mut spawner = TestSpawner::with_downlinks();
-    let modified = run_handler_with_futures_inner(agent, handler, &mut spawner).await;
+    let mut command_buffer = BytesMut::new();
+    let modified =
+        run_handler_with_futures_inner(agent, handler, &mut spawner, &mut command_buffer).await;
     (modified, spawner.take_downlinks())
 }
 
 pub async fn run_handler_with_futures<A, H: EventHandler<A>>(agent: &A, handler: H) -> HashSet<u64>
 where
     A: AgentSpec,
 {
-    run_handler_with_futures_inner(agent, handler, &mut TestSpawner::default()).await
+    run_handler_with_futures_inner(
+        agent,
+        handler,
+        &mut TestSpawner::default(),
+        &mut BytesMut::new(),
+    )
+    .await
+}
+
+pub async fn run_handler_with_futures_and_cmds<A, H: EventHandler<A>>(
+    agent: &A,
+    handler: H,
+    command_buffer: &mut BytesMut,
+) -> HashSet<u64>
+where
+    A: AgentSpec,
+{
+    run_handler_with_futures_inner(agent, handler, &mut TestSpawner::default(), command_buffer)
+        .await
 }
 
 async fn run_handler_with_futures_inner<A, H: EventHandler<A>>(
     agent: &A,
     handler: H,
     spawner: &mut TestSpawner<A>,
+    command_buffer: &mut BytesMut,
 ) -> HashSet<u64>
 where
     A: AgentSpec,
 {
-    let mut modified = run_handler(agent, spawner, handler);
+    let mut modified = run_handler_with_commands(agent, spawner, handler, command_buffer);
     let mut handlers = vec![];
     let reg = move |req: LaneRequest<A>| {
         let LaneRequest {
@@ -211,10 +232,10 @@ where
 
     while !(handlers.is_empty() && spawner.suspended.is_empty()) {
         let m = if let Some(h) = handlers.pop() {
-            run_handler(agent, spawner, h)
+            run_handler_with_commands(agent, spawner, h, command_buffer)
         } else {
             let h = spawner.suspended.next().await.expect("No handler.");
-            run_handler(agent, spawner, h)
+            run_handler_with_commands(agent, spawner, h, command_buffer)
         };
         modified.extend(m);
         for request in