Added egress connector example.

Author: horned-sphere

Cargo.toml

@@ -46,6 +46,7 @@ members = [
     "example_apps/kafka_ingress_connector",
     "example_apps/kafka_egress_connector",
     "example_apps/mqtt_ingress_connector",
+    "example_apps/mqtt_egress_connector"
 ]
 
 [workspace.package]

example_apps/mqtt_egress_connector/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "mqtt-egress-connector"
+version.workspace = true
+edition.workspace = true
+publish = false
+
+[dependencies]
+swimos = { workspace = true, features = ["server", "agent"] }
+swimos_client = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "fs"] }
+example-util = { path = "../example_util" }
+swimos_form = { workspace = true }
+swimos_connector = { workspace = true }
+swimos_connector_mqtt = { workspace = true, features = ["json"] }
+swimos_recon = { workspace = true }
+clap = { workspace = true, features = ["derive"]}
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
+rand = { workspace = true }
+
+[[bin]]
+name = "mqtt_connector_client"

example_apps/mqtt_egress_connector/src/bin/model/mod.rs

@@ -0,0 +1,68 @@
+// 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::SystemTime;
+
+use rand::Rng;
+use swimos_form::Form;
+
+#[derive(Debug, Clone, Copy, Form)]
+pub struct Status {
+    pub severity: f64,
+}
+
+#[derive(Debug, Clone, Copy, Form)]
+pub struct Run {
+    pub mean_ul_sinr: i32,
+    pub rrc_re_establishment_failures: u32,
+    pub recorded_time: u64,
+}
+
+#[derive(Debug, Clone, Copy, Form)]
+pub struct Event {
+    pub status: Status,
+    #[form(name = "ranLatest")]
+    pub ran_latest: Run,
+}
+
+#[derive(Debug, Clone, Copy, Form)]
+pub struct EventWithKey {
+    pub key: i32,
+    pub payload: Event,
+}
+
+pub fn random_record() -> EventWithKey {
+    let mut rng = rand::thread_rng();
+    let severity = rng.gen_range(0.0..2.0);
+    let mean_ul_sinr = rng.gen_range(1..50);
+    let rrc_re_establishment_failures = rng.gen_range(1u32..10u32);
+
+    let t = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .expect("Time out of range.");
+    let recorded_time = t.as_secs() * 1000 + t.subsec_millis() as u64;
+
+    let status = Status { severity };
+    let ran_latest = Run {
+        mean_ul_sinr,
+        rrc_re_establishment_failures,
+        recorded_time,
+    };
+
+    let key = rng.gen_range(50..4000);
+    EventWithKey {
+        key,
+        payload: Event { status, ran_latest },
+    }
+}

example_apps/mqtt_egress_connector/src/bin/mqtt_connector_client.rs

@@ -0,0 +1,37 @@
+// 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::error::Error;
+use swimos_client::Commander;
+use tokio::time::{sleep, Duration};
+
+mod model;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+
+    let mut commander = Commander::default();
+
+    for _ in 0..10 {
+        let record = model::random_record();
+        commander
+            .send_command(host, "/mqtt", "event", &record)
+            .await?;
+        sleep(Duration::from_secs(5)).await;
+    }
+    commander.close().await;
+
+    Ok(())
+}

example_apps/mqtt_egress_connector/src/main.rs

@@ -0,0 +1,112 @@
+// 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.
+
+//! An example demonstrating an MQTT egress connector.
+//!
+//! Run the server using the following:
+//! ```text
+//! $ cargo run --bin mqtt-egress-connector
+//! ```
+//!
+//! And run the client with the following:
+//! ```text
+//! $ cargo run --bin mqtt_connector_client
+//! ```
+
+use std::{error::Error, str::FromStr, time::Duration};
+
+use clap::Parser;
+use example_util::{example_filter, manage_handle};
+use swimos::{
+    route::{RoutePattern, RouteUri},
+    server::{Server, ServerBuilder},
+};
+use swimos_connector::EgressConnectorModel;
+use swimos_connector_mqtt::{MqttEgressConfiguration, MqttEgressConnector};
+use swimos_recon::parser::parse_recognize;
+
+mod params;
+
+use params::Params;
+use tracing::error;
+use tracing_subscriber::filter::LevelFilter;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let Params {
+        config,
+        enable_logging,
+        broker_url,
+    } = Params::parse();
+    if enable_logging {
+        setup_logging()?;
+    }
+
+    let connector_config = load_config(config, broker_url).await?;
+
+    let route = RoutePattern::parse_str("/mqtt")?;
+
+    let connector_agent = EgressConnectorModel::for_fn(move || {
+        MqttEgressConnector::for_config(connector_config.clone())
+    });
+
+    let server = ServerBuilder::with_plane_name("Example Plane")
+        .set_bind_addr("127.0.0.1:8080".parse()?)
+        .add_route(route, connector_agent)
+        .update_config(|config| {
+            config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
+        })
+        .build()
+        .await?;
+
+    let (task, handle) = server.run();
+
+    let uri = RouteUri::from_str("/mqtt")?;
+
+    let shutdown = async move {
+        if let Err(error) = handle.start_agent(uri).await {
+            error!(error = %error, "Failed to start connector agent.");
+        }
+        manage_handle(handle).await
+    };
+
+    let (_, result) = tokio::join!(shutdown, task);
+
+    result?;
+    println!("Server stopped successfully.");
+    Ok(())
+}
+
+const CONNECTOR_CONFIG: &str = include_str!("mqtt_connector.recon");
+
+async fn load_config(
+    path: Option<String>,
+    broker_url: Option<String>,
+) -> Result<MqttEgressConfiguration, Box<dyn Error + Send + Sync>> {
+    let recon = if let Some(path) = path {
+        tokio::fs::read_to_string(path).await?
+    } else if let Some(url) = broker_url {
+        CONNECTOR_CONFIG.replace("##BROKER##", &url)
+    } else {
+        CONNECTOR_CONFIG.replace("##BROKER##", "mqtt://localhost:1883?client_id=example")
+    };
+    let config = parse_recognize::<MqttEgressConfiguration>(recon.as_str(), true)?;
+    Ok(config)
+}
+
+pub fn setup_logging() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let filter = example_filter()?.add_directive(LevelFilter::INFO.into());
+    tracing_subscriber::fmt().with_env_filter(filter).init();
+    Ok(())
+}

example_apps/mqtt_egress_connector/src/mqtt_connector.recon

@@ -0,0 +1,18 @@
+@mqtt {
+    url: "##BROKER##",
+    fixed_topic: "example/topic",
+    value_lanes: {
+        @EgressLaneSpec {
+            name: event,
+            extractor: @ExtractionSpec {
+                topic_specifier: @Fixed,
+                payload_selector: "$value"
+            }
+        }
+    },
+    map_lanes: {},
+    value_downlinks: {},
+    map_downlinks: {},
+    payload_serializer: @Json,
+    channel_size: 0
+}

example_apps/mqtt_egress_connector/src/params.rs

@@ -0,0 +1,29 @@
+// 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 clap::Parser;
+
+#[derive(Parser)]
+#[command(author, version, about, long_about = None)]
+pub struct Params {
+    /// Specify a Recon configuration file for the connector.
+    #[arg(long)]
+    pub config: Option<String>,
+    #[arg(long, default_value = "false")]
+    /// Specify that logging should be enabled.
+    pub enable_logging: bool,
+    /// Specify the MQTT broker (used if --config is not specified.)
+    #[arg(short, long)]
+    pub broker_url: Option<String>,
+}

example_apps/mqtt_ingress_connector/src/mqtt_connector.recon

@@ -1,18 +1,14 @@
 @mqtt {
     url: "mqtt://localhost:1883?client_id=example",
-    fixed_topic: "example/topic",
     value_lanes: {
-        @EgressLaneSpec {
+        @ValueLaneSpec {
             name: value,
-            extractor: @ExtractionSpec {
-                topic_specifier: @Fixed,
-                payload_selector: "$value"
-            }
+            selector: "$payload",
+            required: true
         }
     },
     map_lanes: {},
-    value_downlinks: {},
-    map_downlinks: {},
-    payload_serializer: @Json,
-    channel_size: 0
+    relays: {},
+    payload_deserializer: @Json,
+    subscription: @topic("example/topic")
 }