Fluvio ingress connector

Cargo.toml

@@ -104,6 +104,7 @@ swimos_rtree = { path = "swimos_utilities/swimos_rtree", version = "0.1.1" }
 swimos_sync = { path = "swimos_utilities/swimos_sync", version = "0.1.1" }
 swimos_time = { path = "swimos_utilities/swimos_time", version = "0.1.1" }
 swimos_encoding = { path = "swimos_utilities/swimos_encoding", version = "0.1.1" }
+swimos_connector_util = { path = "server/swimos_connector_util", version = "0.1.1" }
 
 bytes = "1.3"
 tokio = "1.22"
@@ -182,4 +183,5 @@ http-body-util = "0.1.2"
 hyper-util = "0.1.5"
 rdkafka = "0.36"
 apache-avro = "0.17.0"
-time = "0.3.36"
+time = "0.3.36"
+fluvio = "0.23.2"

example_apps/kafka_ingress_connector/src/main.rs

@@ -34,7 +34,6 @@ use swimos::{
 };
 use swimos_connector::IngressConnectorModel;
 use swimos_connector_kafka::{KafkaIngressConfiguration, KafkaIngressConnector};
-use swimos_recon::parser::parse_recognize;
 
 mod params;
 
@@ -99,11 +98,10 @@ async fn load_config(
     } else {
         CONNECTOR_CONFIG
     };
-    let config = parse_recognize::<KafkaIngressConfiguration>(recon, true)?;
-    Ok(config)
+    KafkaIngressConfiguration::from_str(recon)
 }
 
-pub fn setup_logging() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+pub fn setup_logging() -> Result<(), Box<dyn Error + Send + Sync>> {
     let filter = example_filter()?.add_directive(LevelFilter::INFO.into());
     tracing_subscriber::fmt().with_env_filter(filter).init();
     Ok(())

server/swimos_connector/Cargo.toml

@@ -8,6 +8,12 @@ license.workspace = true
 repository = "https://github.com/swimos/swim-rust/tree/main/server/swimos_connector"
 homepage.workspace = true
 
+[features]
+default = []
+json = ["dep:serde_json"]
+avro = ["dep:apache-avro", "dep:chrono", "tokio/fs"]
+pubsub = []
+
 [dependencies]
 futures = { workspace = true }
 swimos_utilities = { workspace = true }
@@ -22,11 +28,21 @@ swimos_agent = { workspace = true }
 swimos_agent_protocol = { workspace = true }
 tokio-stream = { workspace = true }
 tracing = { workspace = true }
-frunk = { workspace = true }
 uuid = { workspace = true }
 thiserror = { workspace = true }
 bitflags = { workspace = true }
+chrono = { workspace = true, optional = true }
+serde_json = { workspace = true, optional = true }
+apache-avro = { workspace = true, optional = true }
+regex = { workspace = true }
+frunk = { workspace = true }
+nom = { workspace = true }
+nom_locate = { workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["rt", "test-util", "time"] }
 parking_lot = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+uuid = { workspace = true, features = ["v4"] }
+swimos_connector_util = { workspace = true }

server/swimos_connector_kafka/src/config/format.rs → server/swimos_connector/src/config/format.rs

@@ -14,21 +14,19 @@
 
 use swimos_form::Form;
 
-use crate::{
-    deser::{
-        BoxMessageDeserializer, BytesDeserializer, F32Deserializer, F64Deserializer,
-        I32Deserializer, I64Deserializer, MessageDeserializer, ReconDeserializer,
-        StringDeserializer, U32Deserializer, U64Deserializer, UuidDeserializer,
-    },
-    ser::{
-        BytesSerializer, F32Serializer, F64Serializer, I32Serializer, I64Serializer,
-        MessageSerializer, ReconSerializer, SharedMessageSerializer, StringSerializer,
-        U32Serializer, U64Serializer, UuidSerializer,
-    },
-    Endianness, LoadError,
+use crate::deser::{
+    BoxMessageDeserializer, BytesDeserializer, Endianness, F32Deserializer, F64Deserializer,
+    I32Deserializer, I64Deserializer, MessageDeserializer, ReconDeserializer, StringDeserializer,
+    U32Deserializer, U64Deserializer, UuidDeserializer,
+};
+use crate::ser::{
+    BytesSerializer, F32Serializer, F64Serializer, I32Serializer, I64Serializer, MessageSerializer,
+    ReconSerializer, SharedMessageSerializer, StringSerializer, U32Serializer, U64Serializer,
+    UuidSerializer,
 };
+use crate::LoadError;
 
-/// Supported deserialization formats to use to interpret a component of a Kafka message.
+/// Supported deserialization formats to use to interpret a component of a message.
 #[derive(Clone, Form, Debug, Default, PartialEq, Eq)]
 pub enum DataFormat {
     #[default]

server/swimos_connector/src/config/ingress.rs

@@ -0,0 +1,174 @@
+use crate::selector::{BadSelector, PubSubSelector, Relay, Relays};
+use swimos_form::Form;
+
+/// Specification of a value lane for the connector.
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
+#[form(tag = "ValueLaneSpec")]
+pub struct IngressValueLaneSpec {
+    /// A name to use for the lane. If not specified, the connector will attempt to infer one from the selector.
+    pub name: Option<String>,
+    /// String representation of a selector to extract values for the lane from messages.
+    pub selector: String,
+    /// Whether the lane is required. If this is `true` and the selector returns nothing for a Message, the
+    /// connector will fail with an error.
+    pub required: bool,
+}
+
+impl IngressValueLaneSpec {
+    /// # Arguments
+    /// * `name` - A name to use for the lane. If not specified the connector will attempt to infer a name from the selector.
+    /// * `selector` - String representation of the selector to extract values from the message.
+    /// * `required` - Whether the lane is required. If this is `true` and the selector returns nothing for a Message, the
+    ///   connector will fail with an error.
+    pub fn new<S: Into<String>>(name: Option<S>, selector: S, required: bool) -> Self {
+        IngressValueLaneSpec {
+            name: name.map(Into::into),
+            selector: selector.into(),
+            required,
+        }
+    }
+}
+
+/// Specification of a value lane for the connector.
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
+#[form(tag = "MapLaneSpec")]
+pub struct IngressMapLaneSpec {
+    /// The name of the lane.
+    pub name: String,
+    /// String representation of a selector to extract the map keys from the messages.
+    pub key_selector: String,
+    /// String representation of a selector to extract the map values from the messages.
+    pub value_selector: String,
+    /// Whether to remove an entry from the map if the value selector does not return a value. Otherwise, missing
+    /// values will be treated as a failed extraction from the message.
+    pub remove_when_no_value: bool,
+    /// Whether the lane is required. If this is `true` and the selector returns nothing for a Message, the
+    /// connector will fail with an error.
+    pub required: bool,
+}
+
+impl IngressMapLaneSpec {
+    /// # Arguments
+    /// * `name` - The name of the lane.
+    /// * `key_selector` - String representation of a selector to extract the map keys from the messages.
+    /// * `value_selector` - String representation of a selector to extract the map values from the messages.
+    /// * `remove_when_no_value` - Whether to remove an entry from the map if the value selector does not return a value. Otherwise, missing
+    ///   values will be treated as a failed extraction from the message.
+    /// * `required` - Whether the lane is required. If this is `true` and the selector returns nothing for a Message, the
+    ///   connector will fail with an error.
+    pub fn new<S: Into<String>>(
+        name: S,
+        key_selector: S,
+        value_selector: S,
+        remove_when_no_value: bool,
+        required: bool,
+    ) -> Self {
+        IngressMapLaneSpec {
+            name: name.into(),
+            key_selector: key_selector.into(),
+            value_selector: value_selector.into(),
+            remove_when_no_value,
+            required,
+        }
+    }
+}
+
+/// Specification of a value relay for the connector.
+#[cfg(feature = "pubsub")]
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
+#[form(tag = "ValueRelaySpec")]
+pub struct PubSubValueRelaySpecification {
+    /// A node URI selector. See [`crate::selector::NodeSelector`] for more information.
+    pub node: String,
+    /// A lane URI selector. See [`crate::selector::LaneSelector`] for more information.
+    pub lane: String,
+    /// A payload URI selector. See [`crate::selector::RelayPayloadSelector::value`] for more information.
+    pub payload: String,
+    /// Whether the payload selector must yield a value. If it does not, then the selector will
+    /// yield an error.
+    pub required: bool,
+}
+
+/// Specification of a map relay for the connector.
+#[cfg(feature = "pubsub")]
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
+#[form(tag = "MapRelaySpec")]
+pub struct PubSubMapRelaySpecification {
+    /// A node URI selector. See [`crate::selector::NodeSelector`] for more information.
+    pub node: String,
+    /// A lane URI selector. See [`crate::selector::LaneSelector`] for more information.
+    pub lane: String,
+    /// A payload URI selector. See [`crate::selector::RelayPayloadSelector::map`] for more information.
+    pub key: String,
+    /// A payload URI selector. See [`crate::selector::RelayPayloadSelector::map`] for more information.
+    pub value: String,
+    /// Whether the payload selector must yield a value. If it does not, then the selector will
+    /// yield an error.
+    pub required: bool,
+    /// If the value selector fails to select, then it will emit a map remove command to remove the
+    /// corresponding entry.
+    pub remove_when_no_value: bool,
+}
+
+/// Specification of a relay for the connector.
+#[cfg(feature = "pubsub")]
+#[derive(Clone, Debug, Form, PartialEq, Eq)]
+pub enum PubSubRelaySpecification {
+    /// Specification of a value relay for the connector.
+    Value(PubSubValueRelaySpecification),
+    /// Specification of a map relay for the connector.
+    Map(PubSubMapRelaySpecification),
+}
+
+#[cfg(feature = "pubsub")]
+impl TryFrom<Vec<PubSubRelaySpecification>> for Relays<PubSubSelector> {
+    type Error = BadSelector;
+
+    fn try_from(value: Vec<PubSubRelaySpecification>) -> Result<Self, Self::Error> {
+        use crate::selector::{
+            parse_lane_selector, parse_map_selector, parse_node_selector, parse_value_selector,
+        };
+
+        let mut chain = Vec::with_capacity(value.len());
+
+        for spec in value {
+            match spec {
+                PubSubRelaySpecification::Value(PubSubValueRelaySpecification {
+                    node,
+                    lane,
+                    payload,
+                    required,
+                }) => {
+                    let relay = Relay::new(
+                        parse_node_selector(node.as_str())?,
+                        parse_lane_selector(lane.as_str())?,
+                        parse_value_selector(payload.as_str(), required)?,
+                    );
+                    chain.push(relay);
+                }
+                PubSubRelaySpecification::Map(PubSubMapRelaySpecification {
+                    node,
+                    lane,
+                    key,
+                    value,
+                    required,
+                    remove_when_no_value,
+                }) => {
+                    let relay = Relay::new(
+                        parse_node_selector(node.as_str())?,
+                        parse_lane_selector(lane.as_str())?,
+                        parse_map_selector(
+                            key.as_str(),
+                            value.as_str(),
+                            required,
+                            remove_when_no_value,
+                        )?,
+                    );
+                    chain.push(relay);
+                }
+            }
+        }
+
+        Ok(Relays::new(chain))
+    }
+}

server/swimos_connector/src/config/mod.rs

@@ -0,0 +1,4 @@
+pub mod format;
+mod ingress;
+
+pub use ingress::*;