Merge branch 'kafka-egress-example' into connector-lanes

Author: horned-sphere

example_apps/aggregations/src/car.rs

@@ -54,15 +54,15 @@ impl CarLifecycle {
                         // deregister this car with its current area
                         let deregister_handler = context.send_command(
                             None,
-                            format!("/area/{old_area}"),
-                            "registrations".to_string(),
+                            &format!("/area/{old_area}"),
+                            "registrations",
                             Action::Deregister(car_id),
                         );
                         // register this car with its new assigned area
                         let register_handler = context.send_command(
                             None,
-                            format!("/area/{new_area}"),
-                            "registrations".to_string(),
+                            &format!("/area/{new_area}"),
+                            "registrations",
                             Action::Register(car_id),
                         );
 

example_apps/join_value/src/room.rs

@@ -45,8 +45,8 @@ impl RoomLifecycle {
 
                             context.send_command(
                                 None,
-                                format!("/buildings/{building_name}"),
-                                "register_room".to_string(),
+                                &format!("/buildings/{building_name}"),
+                                "register_room",
                                 room_id_str,
                             )
                         })

example_apps/transit/src/agents/agency.rs

@@ -80,12 +80,7 @@ impl AgencyLifecycle {
         let state_uri = self.agency.state_uri();
 
         //Associate this agency with the state that contains it.
-        let add_to_state = context.send_command(
-            None,
-            state_uri,
-            "addAgency".to_string(),
-            self.agency.clone(),
-        );
+        let add_to_state = context.send_command(None, &state_uri, "addAgency", self.agency.clone());
         context
             .get_agent_uri()
             .and_then(move |uri| {
@@ -254,8 +249,7 @@ fn process_new_vehicles(
     let additions = new_vehicles
         .into_iter()
         .map(move |(k, v)| {
-            let to_vehicle_agent =
-                context.send_command(None, v.uri.clone(), "addVehicle".to_string(), v.clone());
+            let to_vehicle_agent = context.send_command(None, &v.uri, "addVehicle", v.clone());
             let add_vehicle = context.update(AgencyAgent::VEHICLES, k, v);
             to_vehicle_agent.followed_by(add_vehicle)
         })

example_apps/transit/src/agents/state.rs

@@ -89,8 +89,7 @@ impl StateLifecycle {
             &agency_uri,
             "speed",
         );
-        let add_to_country =
-            context.send_command(None, country_uri, "addAgency".to_string(), agency.clone());
+        let add_to_country = context.send_command(None, &country_uri, "addAgency", agency.clone());
         log_uri
             .followed_by(link_count)
             .followed_by(link_speed)

runtime/swimos_rocks_store/src/server/mod.rs

@@ -279,7 +279,7 @@ impl ServerPersistence for RocksServerPersistence {
 }
 
 /// Open a RocksDB persistence store from a path in the local filesystem. If the specified database does
-/// not exist it will be crated.
+/// not exist it will be created.
 ///
 /// # Arguments
 /// * `path` - The filesystem path to the database. If none is specified, a new database will be created in a temporary directory.

server/swimos_agent/src/agent_lifecycle/utility/downlink_builder/event.rs

@@ -43,6 +43,7 @@ pub struct StatelessEventDownlinkBuilder<
     address: Address<Text>,
     config: SimpleDownlinkConfig,
     inner: LC,
+    // This determines whether then downlink reports a kind of Event or MapEvent and makes no functional difference.
     map_events: bool,
 }
 
@@ -74,7 +75,9 @@ impl<Context, T> StatelessEventDownlinkBuilder<Context, T> {
 }
 
 impl<Context, K, V> StatelessEventDownlinkBuilder<Context, MapMessage<K, V>> {
-    pub fn new_map(address: Address<Text>, config: SimpleDownlinkConfig) -> Self {
+    // Creates a lifecycle for an event downlink that consumes events from a remote map lane. This alters
+    // the type reported by the downlink from Event to MapEvent and makes not function difference.
+    pub(crate) fn new_map(address: Address<Text>, config: SimpleDownlinkConfig) -> Self {
         StatelessEventDownlinkBuilder {
             _type: PhantomData,
             address,
@@ -97,18 +100,6 @@ impl<Context, T, State> StatefulEventDownlinkBuilder<Context, T, State> {
     }
 }
 
-impl<Context, K, V, State> StatefulEventDownlinkBuilder<Context, MapMessage<K, V>, State> {
-    pub fn new_map(address: Address<Text>, config: SimpleDownlinkConfig, state: State) -> Self {
-        StatefulEventDownlinkBuilder {
-            _type: PhantomData,
-            address,
-            config,
-            inner: StatefulEventDownlinkLifecycle::new(state),
-            map_events: false,
-        }
-    }
-}
-
 impl<Context, T, LC> StatelessEventDownlinkBuilder<Context, T, LC>
 where
     LC: StatelessEventLifecycle<Context, T>,

server/swimos_agent/src/agent_lifecycle/utility/mod.rs

@@ -24,7 +24,6 @@ use futures::{Future, FutureExt, Stream, StreamExt};
 use swimos_agent_protocol::MapMessage;
 use swimos_api::agent::WarpLaneKind;
 use swimos_api::error::LaneSpawnError;
-use swimos_model::Text;
 use tokio::time::Instant;
 
 use swimos_api::address::Address;
@@ -130,18 +129,17 @@ impl<Agent: 'static> HandlerContext<Agent> {
     /// * `node` - The target node hosting the lane.
     /// * `lane` - The name of the target lane.
     /// * `command` - The value to send.
-    pub fn send_command<'a, S, T>(
+    pub fn send_command<'a, T>(
         &self,
-        host: Option<S>,
-        node: S,
-        lane: S,
+        host: Option<&str>,
+        node: &str,
+        lane: &str,
         command: T,
     ) -> impl EventHandler<Agent> + 'a
     where
-        S: AsRef<str> + 'a,
         T: StructuralWritable + 'a,
     {
-        let addr = Address::new(host, node, lane);
+        let addr = Address::text(host, node, lane);
         SendCommand::new(addr, command, true)
     }
 
@@ -962,11 +960,11 @@ impl<Agent: 'static> HandlerContext<Agent> {
     /// * `lane` - The name of the target lane.
     pub fn create_commander(
         &self,
-        host: Option<impl Into<Text>>,
-        node: impl Into<Text>,
-        lane: impl Into<Text>,
+        host: Option<&str>,
+        node: &str,
+        lane: &str,
     ) -> impl HandlerAction<Agent, Completion = Commander<Agent>> + 'static {
-        let address = Address::new(host.map(Into::into), node.into(), lane.into());
+        let address = Address::text(host, node, lane);
         RegisterCommander::new(address)
     }
 

server/swimos_agent/src/agent_model/downlink/mod.rs

@@ -83,6 +83,12 @@ impl<T, LC> OpenValueDownlinkAction<T, LC> {
 }
 
 impl<T, LC> OpenEventDownlinkAction<T, LC> {
+    /// # Arguments
+    /// * `address` - The address of the remote lane to link to.
+    /// * `lifecycle` - The lifecycle events associated with the downlink.
+    /// * `config` - Runtime configuration parameters for the downlink.
+    /// * `map_events` - Determines whether the downlink will report [`DownlinkKind::Event`] or
+    ///   [`DownlinkKind::MapEvent`] as its kind (and makes no functional difference).
     pub fn new(
         address: Address<Text>,
         lifecycle: LC,

server/swimos_agent/src/agent_model/tests/external_links/commander_lifecycle.rs

@@ -72,8 +72,8 @@ impl OnStart<EmptyAgent> for CommanderLifecycle {
             context.stop()
         });
 
-        let Address { host, node, lane } = &self.address;
-        let create = context.create_commander(host.as_ref(), node, lane);
+        let Address { host, node, lane } = self.address.borrow_parts();
+        let create = context.create_commander(host, node, lane);
 
         let commands = create.and_then(move |commander: Commander<EmptyAgent>| {
             commander

server/swimos_connector/src/connector/egress/mod.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{collections::HashMap, error::Error, time::Duration};
+use std::{collections::HashMap, time::Duration};
 
 use swimos_api::{address::Address, agent::WarpLaneKind};
 use swimos_model::Value;
@@ -26,28 +26,28 @@ use super::{BaseConnector, ConnectorFuture};
 /// lanes, by default, but allows for them to be added dynamically by the lifecycle. The lanes that a connector
 /// registers can be derived from static configuration or inferred from the external data source itself. Currently,
 /// it is only possible to register dynamic lanes in the initialization phase of the agent (during the `on_start`
-/// event). This restriction should be relaxed in the future.
+/// event). This restriction may be relaxed in the future.
 ///
 /// When the connector starts, it will open some number of value and map lanes. Additionally, a number of event and
 /// map-event downlinks may be opened to remote lanes on other agents. Each time a changes is made to one of the
-/// lanes or an update is received on a downlinks, a message will be sent on a sender, crated by the
+/// lanes or an update is received on a downlinks, a message will be sent on a sender, created by the
 /// [`EgressConnector::make_sender`] method.
 ///
 /// Note that the sender must implement [`Clone`] so that it can be shared between the agent's lanes and the
 /// downlinks.
 pub trait EgressConnector: BaseConnector {
     /// The type of the errors produced by the connector.
-    type SendError: Error + Send + 'static;
+    type Error: std::error::Error + Send + 'static;
 
     /// The type of the sender created by this connector.
-    type Sender: EgressConnectorSender<Self::SendError> + 'static;
+    type Sender: EgressConnectorSender<Self::Error> + 'static;
 
     /// Open the lanes and downlinks required by the connector. This is called during the agent's `on_start`
     /// event.
     ///
     /// # Arguments
     /// * `context` - The connector makes calls to the context to request the lanes and downlinks.
-    fn initialize(&self, context: &mut dyn EgressContext) -> Result<(), Self::SendError>;
+    fn initialize(&self, context: &mut dyn EgressContext) -> Result<(), Self::Error>;
 
     /// Create sender for the connector which is used to send messages to the external data sink. This is called
     /// exactly ones during the agent's `on_start` event but must implement [`Clone`] so that copies can be passed
@@ -58,7 +58,7 @@ pub trait EgressConnector: BaseConnector {
     fn make_sender(
         &self,
         agent_params: &HashMap<String, String>,
-    ) -> Result<Self::Sender, Self::SendError>;
+    ) -> Result<Self::Sender, Self::Error>;
 }
 
 /// Possible results of sending a message to the external sink.