Merge pull request #697 from swimos/connector-api2

Adds the first version of the connector API and Kafka connector.

Author: horned-sphere

.github/workflows/ci.yml

@@ -126,9 +126,9 @@ jobs:
       - name: Checkout repository
         uses: actions/checkout@v2
 
-      - name: Install Clang
-        # Required for rocksdb
-        run: apt-get update && apt-get install -y llvm llvm-dev clang
+      - name: Install Clang & cmake
+        # Required for rocksdb and libdkafka
+        run: apt-get update && apt-get install -y llvm llvm-dev clang cmake
 
       - name: Set libclang path
         run: echo "LIBCLANG_PATH=$(llvm-config --libdir)" >> $GITHUB_ENV

Cargo.toml

@@ -81,6 +81,8 @@ swimos_agent = { path = "server/swimos_agent", version = "0.1.0" }
 swimos_agent_derive = { path = "server/swimos_agent_derive", version = "0.1.0" }
 swimos_introspection = { path = "server/swimos_introspection", version = "0.1.0" }
 swimos_server_app = { path = "server/swimos_server_app", version = "0.1.0" }
+swimos_connector = { path = "server/swimos_connector", version = "0.1.0"}
+swimos_connector_kafka = { path = "server/swimos_connector_kafka", version = "0.1.0"}
 swimos = { path = "swimos", version = "0.1.0" }
 swimos_client = { path = "swimos_client", version = "0.1.0" }
 swimos_downlink = { path = "swimos_downlink", version = "0.1.0" }
@@ -174,4 +176,7 @@ waker-fn = "1.1.0"
 num = "0.4"
 smol_str = "0.2.0"
 http-body-util = "0.1.2"
-hyper-util = "0.1.5"
+hyper-util = "0.1.5"
+rdkafka = "0.25"
+apache-avro = "0.16.0"
+time = "0.3.36"

api/formats/swimos_recon/src/recon_parser/error.rs

@@ -29,7 +29,7 @@ pub enum ParseError {
         line: u32,
         column: usize,
     },
-    /// The parsed strucuture was not valid for the target type.
+    /// The parsed structure was not valid for the target type.
     Structure(ReadError),
     /// The parser produced an invalid stream of events. This likely indicates
     /// a bug in the parser.

api/swimos_api/src/error/mod.rs

@@ -24,6 +24,7 @@ use swimos_utilities::{errors::Recoverable, routing::UnapplyError, trigger::prom
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot, watch};
 
+use crate::agent::WarpLaneKind;
 use crate::{address::RelativeAddress, agent::StoreKind};
 
 mod introspection;
@@ -127,6 +128,34 @@ pub enum OpenStoreError {
     },
 }
 
+/// Error type for requests to a running agent to register a new lane.
+#[derive(Clone, Debug, Error, PartialEq, Eq)]
+pub enum DynamicRegistrationError {
+    #[error("This agent does not support dynamically registered items.")]
+    DynamicRegistrationsNotSupported,
+    #[error("This agent only supports dynamic registration during initialization.")]
+    AfterInitialization,
+    #[error("The requested item name '{0}' is already in use.")]
+    DuplicateName(String),
+    #[error("This agent does not support dynamically adding lanes of type: {0}")]
+    LaneKindUnsupported(WarpLaneKind),
+    #[error("This agent does not support dynamically adding stores of type: {0}")]
+    StoreKindUnsupported(StoreKind),
+    #[error("This agent does not support dynamically adding HTTP lanes.")]
+    HttpLanesUnsupported,
+}
+
+/// Error type for an operation to add a new lane to a running agent.
+#[derive(Clone, Debug, Error, PartialEq, Eq)]
+pub enum LaneSpawnError {
+    /// The agent runtime stopped before the request could be completed,
+    #[error(transparent)]
+    Runtime(#[from] AgentRuntimeError),
+    /// The agent refused to register the lane.
+    #[error(transparent)]
+    Registration(#[from] DynamicRegistrationError),
+}
+
 impl<T> From<mpsc::error::SendError<T>> for AgentRuntimeError {
     fn from(_: mpsc::error::SendError<T>) -> Self {
         AgentRuntimeError::Terminated
@@ -235,6 +264,8 @@ pub enum AgentInitError {
     UserCodeError(Box<dyn std::error::Error + Send>),
     #[error("Initializing the state of an agent lane failed: {0}")]
     LaneInitializationFailure(FrameIoError),
+    #[error("Attempting to dynamically register a lane failed: {0}")]
+    RegistrationFailure(#[from] DynamicRegistrationError),
     #[error(
         "Requested a store of kind {requested} for item {name} but store was of kind {actual}."
     )]

example_apps/console/Cargo.toml

@@ -22,6 +22,7 @@ console-views = { path = "console_views" }
 http = { workspace = true }
 rand = { workspace = true }
 duration-str = { workspace = true }
+time = { workspace = true }
 thiserror = { workspace = true }
 
 [target.'cfg(windows)'.dependencies]

server/swimos_agent/src/agent_lifecycle/on_start.rs

@@ -20,11 +20,7 @@ use super::utility::HandlerContext;
 
 /// Lifecycle event for the `on_start` event of an agent.
 pub trait OnStart<Context>: Send {
-    type OnStartHandler<'a>: EventHandler<Context> + 'a
-    where
-        Self: 'a;
-
-    fn on_start(&self) -> Self::OnStartHandler<'_>;
+    fn on_start(&self) -> impl EventHandler<Context> + '_;
 }
 
 /// Lifecycle event for the `on_start` event of an agent where the event handler
@@ -46,10 +42,8 @@ pub trait OnStartShared<Context, Shared>: Send {
 }
 
 impl<Context> OnStart<Context> for NoHandler {
-    type OnStartHandler<'a> = UnitHandler where Self: 'a;
-
-    fn on_start(&self) -> Self::OnStartHandler<'_> {
-        Default::default()
+    fn on_start(&self) -> impl EventHandler<Context> + '_ {
+        UnitHandler::default()
     }
 }
 
@@ -58,11 +52,7 @@ where
     F: Fn() -> H + Send,
     H: EventHandler<Context> + 'static,
 {
-    type OnStartHandler<'a> = H
-    where
-        Self: 'a;
-
-    fn on_start(&self) -> Self::OnStartHandler<'_> {
+    fn on_start(&self) -> impl EventHandler<Context> + '_ {
         let FnHandler(f) = self;
         f()
     }

server/swimos_agent/src/agent_lifecycle/on_stop.rs

@@ -20,11 +20,7 @@ use super::utility::HandlerContext;
 
 /// Lifecycle event for the `on_stop` event of an agent.
 pub trait OnStop<Context>: Send {
-    type OnStopHandler<'a>: EventHandler<Context> + 'a
-    where
-        Self: 'a;
-
-    fn on_stop(&self) -> Self::OnStopHandler<'_>;
+    fn on_stop(&self) -> impl EventHandler<Context> + '_;
 }
 
 /// Lifecycle event for the `on_stop` event of an agent where the event handler
@@ -46,12 +42,8 @@ pub trait OnStopShared<Context, Shared>: Send {
 }
 
 impl<Context> OnStop<Context> for NoHandler {
-    type OnStopHandler<'a> = UnitHandler
-    where
-        Self: 'a;
-
-    fn on_stop(&self) -> Self::OnStopHandler<'_> {
-        Default::default()
+    fn on_stop(&self) -> impl EventHandler<Context> + '_ {
+        UnitHandler::default()
     }
 }
 
@@ -75,11 +67,7 @@ where
     F: Fn() -> H + Send,
     H: EventHandler<Context> + 'static,
 {
-    type OnStopHandler<'a> = H
-    where
-        Self: 'a;
-
-    fn on_stop(&self) -> Self::OnStopHandler<'_> {
+    fn on_stop(&self) -> impl EventHandler<Context> + '_ {
         let FnHandler(f) = self;
         f()
     }

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

@@ -15,7 +15,11 @@
 use static_assertions::assert_impl_all;
 use swimos_utilities::handlers::{FnHandler, NoHandler};
 
-use crate::{agent_lifecycle::AgentLifecycle, event_handler::ActionContext, meta::AgentMetadata};
+use crate::{
+    agent_lifecycle::AgentLifecycle,
+    event_handler::{ActionContext, EventHandler},
+    meta::AgentMetadata,
+};
 
 use super::{
     item_event::{ItemEvent, ItemEventShared},
@@ -110,9 +114,7 @@ where
     FStop: Send,
     ItemEv: Send,
 {
-    type OnStartHandler<'a> = FStart::OnStartHandler<'a> where Self: 'a;
-
-    fn on_start(&self) -> Self::OnStartHandler<'_> {
+    fn on_start(&self) -> impl EventHandler<Context> + '_ {
         let StatefulAgentLifecycle {
             state,
             handler_context,
@@ -132,11 +134,7 @@ where
     FStart: Send,
     ItemEv: Send,
 {
-    type OnStopHandler<'a> = FStop::OnStopHandler<'a>
-    where
-        Self: 'a;
-
-    fn on_stop(&self) -> Self::OnStopHandler<'_> {
+    fn on_stop(&self) -> impl EventHandler<Context> + '_ {
         let StatefulAgentLifecycle {
             state,
             handler_context,

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

@@ -21,6 +21,8 @@ use std::{collections::HashMap, marker::PhantomData};
 
 use futures::stream::unfold;
 use futures::{Future, FutureExt, Stream, StreamExt};
+use swimos_api::agent::WarpLaneKind;
+use swimos_api::error::LaneSpawnError;
 use tokio::time::Instant;
 
 use swimos_api::address::Address;
@@ -37,7 +39,7 @@ use crate::config::{MapDownlinkConfig, SimpleDownlinkConfig};
 use crate::downlink_lifecycle::ValueDownlinkLifecycle;
 use crate::downlink_lifecycle::{EventDownlinkLifecycle, MapDownlinkLifecycle};
 use crate::event_handler::{
-    run_after, run_schedule, run_schedule_async, ConstHandler, EventHandler, GetParameter,
+    run_after, run_schedule, run_schedule_async, ConstHandler, EventHandler, Fail, GetParameter,
     HandlerActionExt, SendCommand, Sequentially, Stop, Suspend, UnitHandler,
 };
 use crate::event_handler::{GetAgentUri, HandlerAction, SideEffect};
@@ -51,7 +53,7 @@ use crate::lanes::demand_map::CueKey;
 use crate::lanes::join_map::JoinMapAddDownlink;
 use crate::lanes::join_value::{JoinValueAddDownlink, JoinValueLane};
 use crate::lanes::supply::{Supply, SupplyLane};
-use crate::lanes::{DemandMapLane, JoinMapLane};
+use crate::lanes::{DemandMapLane, JoinMapLane, OpenLane};
 
 pub use self::downlink_builder::event::{
     StatefulEventDownlinkBuilder, StatelessEventDownlinkBuilder,
@@ -862,6 +864,56 @@ impl<Agent: 'static> HandlerContext<Agent> {
     pub fn stop(&self) -> impl EventHandler<Agent> + Send + 'static {
         HandlerActionExt::<Agent>::discard(Stop)
     }
+
+    /// Create a handler that will fail with the provided error.
+    ///
+    /// # Arguments
+    /// * `error` - The error.
+    pub fn fail<T, E>(&self, error: E) -> impl HandlerAction<Agent, Completion = T> + Send + 'static
+    where
+        T: Send + 'static,
+        E: std::error::Error + Send + 'static,
+    {
+        Fail::<T, E>::new(error)
+    }
+
+    /// Attempt to open a new value lane on this agent. Note that the implementation of the underlying agent
+    /// used must support this.
+    ///
+    /// #Arguments
+    ///
+    /// * `name` - The name of the lane.
+    /// * `on_done` - A callback that will create an event handler that will be executed when the request completes.
+    pub fn open_value_lane<OnDone, H>(
+        &self,
+        name: &str,
+        on_done: OnDone,
+    ) -> impl EventHandler<Agent> + Send + 'static
+    where
+        OnDone: FnOnce(Result<(), LaneSpawnError>) -> H + Send + 'static,
+        H: EventHandler<Agent> + Send + 'static,
+    {
+        OpenLane::new(name.to_string(), WarpLaneKind::Value, on_done)
+    }
+
+    /// Attempt to open a new map lane on this agent. Note that the implementation of the underlying agent
+    /// used must support this.
+    ///
+    /// #Arguments
+    ///
+    /// * `name` - The name of the lane.
+    /// * `on_done` - A callback that will create an event handler that will be executed when the request completes.
+    pub fn open_map_lane<OnDone, H>(
+        &self,
+        name: &str,
+        on_done: OnDone,
+    ) -> impl EventHandler<Agent> + Send + 'static
+    where
+        OnDone: FnOnce(Result<(), LaneSpawnError>) -> H + Send + 'static,
+        H: EventHandler<Agent> + Send + 'static,
+    {
+        OpenLane::new(name.to_string(), WarpLaneKind::Map, on_done)
+    }
 }
 
 /// Context passed to agent methods used to construct lifecycles for [`JoinValueLane`]s.

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

@@ -23,7 +23,7 @@ use swimos_utilities::routing::RouteUri;
 use crate::{
     event_handler::{ActionContext, HandlerAction, LocalBoxEventHandler, StepResult},
     meta::AgentMetadata,
-    test_context::{no_downlink, DummyAgentContext},
+    test_context::{no_downlink, DummyAgentContext, NO_DYN_LANES},
 };
 
 use super::HandlerContext;
@@ -103,6 +103,7 @@ async fn suspend_repeatedly() {
                     &spawner,
                     &DummyAgentContext,
                     &no_downlink,
+                    &NO_DYN_LANES,
                     &mut join_lane_init,
                     &mut ad_hoc_buffer,
                 ),