Merge pull request #698 from swimos/kafka-example

Kafka connector example app.

Author: SirCipher

Cargo.toml

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

example_apps/README.md

@@ -13,6 +13,7 @@ binaries for their corresponding client applications. Each directory contains de
 - [http_lane](http_lane): Application demonstrating HTTP Lanes.
 - [join_map](join_map): Application demonstrating Join Map Lanes.
 - [join_value](join_value): Application demonstrating Join Value Lanes.
+- [kafka_connector](kafka_connector): Application demonstrating consuming data from Apache Kafka.
 - [local_downlink](local_downlink): Application demonstrating Downlinks.
 - [map_downlink](map_downlink): Application demonstrating Map Downlinks.
 - [map_lane](map_lane): Application demonstrating Map Lanes.

example_apps/kafka_connector/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "kafka-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_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"] }
+
+[[bin]]
+name = "kafka_connector_client"

example_apps/kafka_connector/src/bin/kafka_connector_client.rs

@@ -0,0 +1,38 @@
+// 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::{BasicValueDownlinkLifecycle, RemotePath, SwimClientBuilder};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let host = "ws://127.0.0.1:8080";
+    let (client, task) = SwimClientBuilder::default().build().await;
+    let task_handle = tokio::spawn(task);
+
+    let client_handle = client.handle();
+
+    let lifecycle =
+        BasicValueDownlinkLifecycle::default().on_set_blocking(|old_value, new_value| {
+            println!("'{old_value:?}' -> '{new_value:?}'");
+        });
+    let _view = client_handle
+        .value_downlink::<i32>(RemotePath::new(host, "/kafka", "latest_key"))
+        .lifecycle(lifecycle)
+        .open()
+        .await?;
+
+    task_handle.await?;
+    Ok(())
+}

example_apps/kafka_connector/src/kafka_connector.recon

@@ -0,0 +1,30 @@
+@kafka {
+    properties: {
+        "group.id": rust-consumer-test,
+        "message.timeout.ms": "5000",
+        "bootstrap.servers": "datagen.nstream.cloud:9092",
+        "auto.offset.reset": smallest
+    },
+    log_level: @Debug,
+    value_lanes: {
+        @ValueLaneSpec {
+            name: latest_key,
+            selector: "$key",
+            required: true
+        }
+    },
+    map_lanes: {
+        @MapLaneSpec {
+            name: times,
+            key_selector: "$payload.ranLatest.mean_ul_sinr",
+            value_selector: "$payload.ranLatest.recorded_time",
+            remove_when_no_value: false,
+            required: true
+        }
+    },
+    key_deserializer: @Int32(@LittleEndian),
+    payload_deserializer: @Json,
+    topics: {
+        cellular-integer-json
+    }
+}

example_apps/kafka_connector/src/main.rs

@@ -0,0 +1,109 @@
+// 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 Value Lanes.
+//!
+//! Run the server using the following:
+//! ```text
+//! $ cargo run --bin value-lane
+//! ```
+//!
+//! And run the client with the following:
+//! ```text
+//! $ cargo run --bin value_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::ConnectorModel;
+use swimos_connector_kafka::{KafkaConnector, KafkaConnectorConfiguration};
+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,
+    } = Params::parse();
+    if enable_logging {
+        setup_logging()?;
+    }
+
+    let connector_config = load_config(config).await?;
+
+    let route = RoutePattern::parse_str("/kafka")?;
+
+    let connector_agent =
+        ConnectorModel::for_fn(move || KafkaConnector::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>,
+) -> Result<KafkaConnectorConfiguration, Box<dyn Error + Send + Sync>> {
+    let content: String;
+    let recon = if let Some(path) = path {
+        content = tokio::fs::read_to_string(path).await?;
+        &content
+    } else {
+        CONNECTOR_CONFIG
+    };
+    let config = parse_recognize::<KafkaConnectorConfiguration>(recon, 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_connector/src/params.rs

@@ -0,0 +1,26 @@
+// 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,
+}

server/swimos_connector_kafka/src/config.rs

@@ -27,7 +27,7 @@ use crate::{
 };
 
 /// Configuration parameters for the Kafka connector.
-#[derive(Clone, Debug, Form)]
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
 #[form(tag = "kafka")]
 pub struct KafkaConnectorConfiguration {
     /// Properties to configure the Kafka consumer.
@@ -49,7 +49,7 @@ pub struct KafkaConnectorConfiguration {
 }
 
 /// Specification of a value lane for the Kafka connector.
-#[derive(Clone, Debug, Form)]
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
 pub struct ValueLaneSpec {
     /// A name to use for the lane. If not specified, the connector will attempt to infer one from the selector.
     pub name: Option<String>,
@@ -76,7 +76,7 @@ impl ValueLaneSpec {
 }
 
 /// Specification of a value lane for the Kafka connector.
-#[derive(Clone, Debug, Form)]
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
 pub struct MapLaneSpec {
     /// The name of the lane.
     pub name: String,
@@ -119,7 +119,7 @@ impl MapLaneSpec {
 }
 
 /// Supported deserialization formats to use to interpret a component of a Kafka message.
-#[derive(Clone, Form, Debug, Default)]
+#[derive(Clone, Form, Debug, Default, PartialEq, Eq)]
 pub enum DeserializationFormat {
     #[default]
     Bytes,

server/swimos_connector_kafka/src/deser/mod.rs

@@ -155,7 +155,7 @@ impl MessageDeserializer for ReconDeserializer {
 }
 
 /// Endianness for numeric deserializers.
-#[derive(Clone, Copy, Default, Debug, Form)]
+#[derive(Clone, Copy, Default, Debug, Form, PartialEq, Eq)]
 pub enum Endianness {
     #[default]
     LittleEndian,