Merge pull request #719 from swimos/kafka-egress-example

Example app for Kafka egress connector.

Author: SirCipher

Cargo.toml

@@ -43,7 +43,8 @@ members = [
     "example_apps/devguide",
     "example_apps/ripple",
     "example_apps/stocks_simulated",
-    "example_apps/kafka_connector"
+    "example_apps/kafka_ingress_connector",
+    "example_apps/kafka_egress_connector"
 ]
 
 [workspace.package]

example_apps/kafka_egress_connector/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "kafka-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_kafka = { 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 = "kafka_connector_client"

example_apps/kafka_egress_connector/src/bin/kafka_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, "/kafka", "event", &record)
+            .await?;
+        sleep(Duration::from_secs(5)).await;
+    }
+    commander.close().await;
+
+    Ok(())
+}

example_apps/kafka_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/kafka_egress_connector/src/kafka_connector.recon

@@ -0,0 +1,25 @@
+@kafka {
+    properties: {
+        "message.timeout.ms": "5000",
+        "group.id": rust-consumer-test,
+        "bootstrap.servers": "##SERVERS##"
+    },
+    log_level: @Debug,
+    key_serializer: @Int32(@BigEndian),
+    payload_serializer: @Json,
+    fixed_topic: cellular-integer-json,
+    value_lanes: {
+        @LaneSpec {
+            name: event,
+            extractor: @ExtractionSpec {
+                topic_specifier: @Fixed,
+                key_selector: "$value.key",
+                payload_selector: "$value.payload"
+            }
+        }
+    },
+    map_lanes: {},
+    value_downlinks: {},
+    map_downlinks: {},
+    retry_timeout_ms: 5000
+}

example_apps/kafka_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 a Kafka egress connector.
+//!
+//! Run the server using the following:
+//! ```text
+//! $ cargo run --bin kafka-egress-connector
+//! ```
+//!
+//! And run the client with the following:
+//! ```text
+//! $ cargo run --bin kafka_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_kafka::{KafkaEgressConfiguration, KafkaEgressConnector};
+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,
+        bootstrap_servers,
+    } = Params::parse();
+    if enable_logging {
+        setup_logging()?;
+    }
+
+    let connector_config = load_config(config, bootstrap_servers).await?;
+
+    let route = RoutePattern::parse_str("/kafka")?;
+
+    let connector_agent = EgressConnectorModel::for_fn(move || {
+        KafkaEgressConnector::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("/kafka")?;
+
+    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!("kafka_connector.recon");
+
+async fn load_config(
+    path: Option<String>,
+    bootstrap_servers: Option<String>,
+) -> Result<KafkaEgressConfiguration, Box<dyn Error + Send + Sync>> {
+    let recon = if let Some(path) = path {
+        tokio::fs::read_to_string(path).await?
+    } else if let Some(bootstrap) = bootstrap_servers {
+        CONNECTOR_CONFIG.replace("##SERVERS##", &bootstrap)
+    } else {
+        panic!("Either a configuration file or bootstrap servers must be specified.");
+    };
+    let config = parse_recognize::<KafkaEgressConfiguration>(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/kafka_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 Kafka bootstrap servers (used if --config is not specified.)
+    #[arg(short, long)]
+    pub bootstrap_servers: Option<String>,
+}

example_apps/kafka_connector/Cargo.toml renamed to example_apps/kafka_ingress_connector/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "kafka-connector"
+name = "kafka-ingress-connector"
 version.workspace = true
 edition.workspace = true
 publish = false

example_apps/kafka_connector/src/bin/kafka_connector_client.rs renamed to example_apps/kafka_ingress_connector/src/bin/kafka_connector_client.rs

@@ -22,7 +22,7 @@
             required: true
         }
     },
-    key_deserializer: @Int32(@LittleEndian),
+    key_deserializer: @Int32(@BigEndian),
     payload_deserializer: @Json,
     topics: {
         cellular-integer-json