Resolves PR comments

Author: SirCipher

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"

server/swimos_connector/Cargo.toml

@@ -44,4 +44,5 @@ tokio = { workspace = true, features = ["rt", "test-util", "time"] }
 parking_lot = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-uuid = { workspace = true, features = ["v4"] }
+uuid = { workspace = true, features = ["v4"] }
+swimos_connector_util = { workspace = true }

server/swimos_connector/src/config/ingress.rs

@@ -1,4 +1,4 @@
-use crate::selector::{BadSelector, Relay, Relays};
+use crate::selector::{BadSelector, PubSubSelector, Relay, Relays};
 use swimos_form::Form;
 
 /// Specification of a value lane for the connector.
@@ -73,66 +73,97 @@ impl IngressMapLaneSpec {
     }
 }
 
+/// Specification of a value relay for the connector.
+#[cfg(feature = "pubsub")]
 #[derive(Clone, Debug, Form, PartialEq, Eq)]
 #[form(tag = "ValueRelaySpec")]
-pub struct ValueRelaySpecification {
+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 MapRelaySpecification {
+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 RelaySpecification {
-    Value(ValueRelaySpecification),
-    Map(MapRelaySpecification),
+pub enum PubSubRelaySpecification {
+    /// Specification of a value relay for the connector.
+    Value(PubSubValueRelaySpecification),
+    /// Specification of a map relay for the connector.
+    Map(PubSubMapRelaySpecification),
 }
 
-impl TryFrom<Vec<RelaySpecification>> for Relays {
+#[cfg(feature = "pubsub")]
+impl TryFrom<Vec<PubSubRelaySpecification>> for Relays<PubSubSelector> {
     type Error = BadSelector;
 
-    fn try_from(value: Vec<RelaySpecification>) -> Result<Self, Self::Error> {
+    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 {
-                RelaySpecification::Value(ValueRelaySpecification {
+                PubSubRelaySpecification::Value(PubSubValueRelaySpecification {
                     node,
                     lane,
                     payload,
                     required,
                 }) => {
-                    let relay =
-                        Relay::value(node.as_str(), lane.as_str(), payload.as_str(), 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);
                 }
-                RelaySpecification::Map(MapRelaySpecification {
+                PubSubRelaySpecification::Map(PubSubMapRelaySpecification {
                     node,
                     lane,
                     key,
                     value,
                     required,
                     remove_when_no_value,
                 }) => {
-                    let relay = Relay::map(
-                        node.as_str(),
-                        lane.as_str(),
-                        key.as_str(),
-                        value.as_str(),
-                        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);
                 }
             }

server/swimos_connector/src/deser/mod.rs

@@ -26,7 +26,6 @@ pub use json::JsonDeserializer;
 #[cfg(feature = "avro")]
 pub use avro::AvroDeserializer;
 
-use std::cell::RefCell;
 use std::error::Error;
 use std::{array::TryFromSliceError, convert::Infallible};
 use swimos_form::Form;
@@ -242,40 +241,37 @@ where
 pub struct Deferred<'a> {
     buf: &'a [u8],
     deser: &'a BoxMessageDeserializer,
-    state: RefCell<Option<Value>>,
+    state: Option<Value>,
 }
 
 impl<'a> Deferred<'a> {
     pub fn new(buf: &'a [u8], deser: &'a BoxMessageDeserializer) -> Deferred<'a> {
         Deferred {
             buf,
             deser,
-            state: RefCell::new(None),
+            state: None,
         }
     }
 
-    pub fn get(&self) -> Result<Value, DeserializationError> {
+    pub fn get(&mut self) -> Result<Value, DeserializationError> {
         let Deferred { buf, deser, state } = self;
-        if let Some(v) = &*state.borrow() {
+        if let Some(v) = state {
             Ok(v.clone())
         } else {
-            let inner = &mut *state.borrow_mut();
-            Ok(inner.insert(deser.deserialize(buf)?).clone())
+            Ok(state.insert(deser.deserialize(buf)?).clone())
         }
     }
 
-    pub fn with<F>(&self, f: F) -> Result<Option<Value>, DeserializationError>
+    pub fn with<F>(&mut self, f: F) -> Result<Option<Value>, DeserializationError>
     where
         F: FnOnce(&Value) -> Option<Value>,
     {
         let Deferred { buf, deser, state } = self;
-        {
-            if let Some(v) = &*state.borrow() {
-                return Ok(f(v));
-            }
+        if let Some(v) = state {
+            Ok(f(v))
+        } else {
+            let val = state.insert(deser.deserialize(buf)?);
+            Ok(f(val))
         }
-        let inner = &mut *state.borrow_mut();
-        let val = inner.insert(deser.deserialize(buf)?);
-        Ok(f(val))
     }
 }

server/swimos_connector/src/ingress/lanes.rs

@@ -13,40 +13,50 @@
 // limitations under the License.
 
 use crate::config::{IngressMapLaneSpec, IngressValueLaneSpec};
-use crate::ingress::check_selectors;
-use crate::selector::{
-    InvalidLanes, MapLaneSelector, PubSubMapLaneSelector, PubSubValueLaneSelector,
-    ValueLaneSelector,
-};
+use crate::selector::{InvalidLaneSpec, InvalidLanes, MapLaneSelector, ValueLaneSelector};
+use std::collections::HashSet;
 
 // Information about the lanes of the connector. These are computed from the configuration in the `on_start` handler
 // and stored in the lifecycle to be used to start the consumer stream.
-#[derive(Debug, Default, Clone)]
-pub struct Lanes {
-    value_lanes: Vec<PubSubValueLaneSelector>,
-    map_lanes: Vec<PubSubMapLaneSelector>,
+#[derive(Debug, Clone)]
+pub struct Lanes<S> {
+    value_lanes: Vec<ValueLaneSelector<S>>,
+    map_lanes: Vec<MapLaneSelector<S, S>>,
 }
 
-impl Lanes {
-    pub fn value_lanes(&self) -> &[PubSubValueLaneSelector] {
+impl<S> Default for Lanes<S> {
+    fn default() -> Self {
+        Lanes {
+            value_lanes: vec![],
+            map_lanes: vec![],
+        }
+    }
+}
+
+impl<S> Lanes<S> {
+    pub fn value_lanes(&self) -> &[ValueLaneSelector<S>] {
         &self.value_lanes
     }
 
-    pub fn map_lanes(&self) -> &[PubSubMapLaneSelector] {
+    pub fn map_lanes(&self) -> &[MapLaneSelector<S, S>] {
         &self.map_lanes
     }
 
-    pub fn try_from_lane_specs(
-        value_lanes: &[IngressValueLaneSpec],
-        map_lanes: &[IngressMapLaneSpec],
-    ) -> Result<Lanes, InvalidLanes> {
+    pub fn try_from_lane_specs<'a>(
+        value_lanes: &'a [IngressValueLaneSpec],
+        map_lanes: &'a [IngressMapLaneSpec],
+    ) -> Result<Lanes<S>, InvalidLanes>
+    where
+        ValueLaneSelector<S>: TryFrom<&'a IngressValueLaneSpec, Error = InvalidLaneSpec>,
+        MapLaneSelector<S, S>: TryFrom<&'a IngressMapLaneSpec, Error = InvalidLaneSpec>,
+    {
         let value_selectors = value_lanes
             .iter()
-            .map(ValueLaneSelector::try_from)
+            .map(ValueLaneSelector::<S>::try_from)
             .collect::<Result<Vec<_>, _>>()?;
         let map_selectors = map_lanes
             .iter()
-            .map(MapLaneSelector::try_from)
+            .map(MapLaneSelector::<S, S>::try_from)
             .collect::<Result<Vec<_>, _>>()?;
         check_selectors(&value_selectors, &map_selectors)?;
         Ok(Lanes {
@@ -55,3 +65,27 @@ impl Lanes {
         })
     }
 }
+
+fn check_selectors<S>(
+    value_selectors: &[ValueLaneSelector<S>],
+    map_selectors: &[MapLaneSelector<S, S>],
+) -> Result<(), InvalidLanes> {
+    let mut names = HashSet::new();
+    for value_selector in value_selectors {
+        let name = value_selector.name();
+        if names.contains(name) {
+            return Err(InvalidLanes::NameCollision(name.to_string()));
+        } else {
+            names.insert(name);
+        }
+    }
+    for map_selector in map_selectors {
+        let name = map_selector.name();
+        if names.contains(name) {
+            return Err(InvalidLanes::NameCollision(name.to_string()));
+        } else {
+            names.insert(name);
+        }
+    }
+    Ok(())
+}