Add trait FlowRegistry

The plan is to have two implementations
- The BaseFlowRegistry which provides no specific support to poll or
  stream sources.
- The ConnectedFlowRegistry which handles sources

Signed-off-by: Didier Wenzek <[email protected]>

Author: didier-wenzek

crates/core/tedge/src/cli/flows/cli.rs

@@ -8,7 +8,8 @@ use anyhow::Context;
 use anyhow::Error;
 use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
-use tedge_flows::flow::Message;
+use tedge_flows::BaseFlowRegistry;
+use tedge_flows::Message;
 use tedge_flows::MessageProcessor;
 
 #[derive(clap::Subcommand, Debug)]
@@ -110,8 +111,10 @@ impl TEdgeFlowsCli {
         config.root_dir().join("flows")
     }
 
-    pub async fn load_flows(flows_dir: &Utf8PathBuf) -> Result<MessageProcessor, Error> {
-        let mut processor = MessageProcessor::try_new(flows_dir)
+    pub async fn load_flows(
+        flows_dir: &Utf8PathBuf,
+    ) -> Result<MessageProcessor<BaseFlowRegistry>, Error> {
+        let mut processor = MessageProcessor::with_base_registry(flows_dir)
             .await
             .with_context(|| format!("loading flows and steps from {flows_dir}"))?;
         processor.load_all_flows().await;
@@ -121,8 +124,8 @@ impl TEdgeFlowsCli {
     pub async fn load_file(
         flows_dir: &Utf8PathBuf,
         path: &Utf8PathBuf,
-    ) -> Result<MessageProcessor, Error> {
-        let mut processor = MessageProcessor::try_new(flows_dir)
+    ) -> Result<MessageProcessor<BaseFlowRegistry>, Error> {
+        let mut processor = MessageProcessor::with_base_registry(flows_dir)
             .await
             .with_context(|| format!("loading flow {path}"))?;
 

crates/core/tedge/src/cli/flows/list.rs

@@ -4,7 +4,8 @@ use crate::log::MaybeFancy;
 use anyhow::Error;
 use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
-use tedge_flows::flow::Flow;
+use tedge_flows::Flow;
+use tedge_flows::FlowRegistryExt;
 
 pub struct ListCommand {
     pub flows_dir: Utf8PathBuf,

crates/core/tedge/src/cli/flows/test.rs

@@ -6,8 +6,12 @@ use base64::prelude::BASE64_STANDARD;
 use base64::prelude::*;
 use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
-use tedge_flows::flow::*;
+use tedge_flows::BaseFlowRegistry;
+use tedge_flows::DateTime;
+use tedge_flows::FlowResult;
+use tedge_flows::Message;
 use tedge_flows::MessageProcessor;
+use tedge_flows::SourceTag;
 use tokio::io::AsyncBufReadExt;
 use tokio::io::BufReader;
 use tokio::io::Stdin;
@@ -61,7 +65,7 @@ impl Command for TestCommand {
 impl TestCommand {
     async fn process(
         &self,
-        processor: &mut MessageProcessor,
+        processor: &mut MessageProcessor<BaseFlowRegistry>,
         mut message: Message,
         timestamp: DateTime,
     ) {
@@ -82,7 +86,12 @@ impl TestCommand {
             .for_each(|msg| self.print_messages(msg))
     }
 
-    async fn tick(&self, processor: &mut MessageProcessor, timestamp: DateTime, now: Instant) {
+    async fn tick(
+        &self,
+        processor: &mut MessageProcessor<BaseFlowRegistry>,
+        timestamp: DateTime,
+        now: Instant,
+    ) {
         processor
             .on_interval(timestamp, now)
             .await

crates/extensions/tedge_flows/src/actor.rs

@@ -4,6 +4,8 @@ use crate::flow::FlowOutput;
 use crate::flow::FlowResult;
 use crate::flow::Message;
 use crate::flow::SourceTag;
+use crate::registry::BaseFlowRegistry;
+use crate::registry::FlowRegistryExt;
 use crate::runtime::MessageProcessor;
 use crate::InputMessage;
 use crate::Tick;
@@ -44,7 +46,7 @@ pub struct FlowsMapper {
     pub(super) watch_request_sender: DynSender<WatchRequest>,
     pub(super) subscriptions: TopicFilter,
     pub(super) watched_commands: HashSet<String>,
-    pub(super) processor: MessageProcessor,
+    pub(super) processor: MessageProcessor<BaseFlowRegistry>,
     pub(super) next_dump: Instant,
 }
 
@@ -174,16 +176,16 @@ impl FlowsMapper {
     async fn notify_flows_status(&mut self) -> Result<(), RuntimeError> {
         let status = "enabled";
         let now = OffsetDateTime::now_utc();
-        for flow in self.processor.flows.keys() {
-            let status = Self::flow_status(flow, status, &now);
+        for flow in self.processor.registry.flows() {
+            let status = Self::flow_status(flow.name(), status, &now);
             self.mqtt_sender.send(status).await?;
         }
         Ok(())
     }
 
     async fn update_flow_status(&mut self, flow: &str) -> Result<(), RuntimeError> {
         let now = OffsetDateTime::now_utc();
-        let status = if self.processor.flows.contains_key(flow) {
+        let status = if self.processor.registry.contains_flow(flow) {
             "updated"
         } else {
             "removed"
@@ -270,7 +272,7 @@ impl FlowsMapper {
         flow_name: String,
         line: String,
     ) -> Result<(), RuntimeError> {
-        if let Some(flow) = self.processor.registry.get(&flow_name) {
+        if let Some(flow) = self.processor.registry.flow(&flow_name) {
             let topic = flow.input.enforced_topic().unwrap_or_default();
             let source = SourceTag::Process {
                 flow: flow_name.clone(),
@@ -286,7 +288,7 @@ impl FlowsMapper {
         flow_name: &str,
         error: FlowError,
     ) -> Result<(), RuntimeError> {
-        let Some((info, flow_error)) = self.processor.registry.get(flow_name).map(|flow| {
+        let Some((info, flow_error)) = self.processor.registry.flow(flow_name).map(|flow| {
             (
                 format!("Reconnecting input: {flow_name}: {}", flow.input),
                 flow.on_error(error),
@@ -299,7 +301,7 @@ impl FlowsMapper {
         let Some(request) = self
             .processor
             .registry
-            .get(flow_name)
+            .flow(flow_name)
             .and_then(|flow| flow.watch_request())
         else {
             return Ok(());
@@ -315,7 +317,7 @@ impl FlowsMapper {
     }
 
     async fn on_process_eos(&mut self, flow_name: &str) -> Result<(), RuntimeError> {
-        if let Some(flow) = self.processor.registry.get(flow_name) {
+        if let Some(flow) = self.processor.registry.flow(flow_name) {
             if let Some(request) = flow.watch_request() {
                 info!(target: "flows", "Reconnecting input: {flow_name}: {}", flow.input);
                 self.watch_request_sender.send(request).await?

crates/extensions/tedge_flows/src/config.rs

@@ -12,10 +12,15 @@ use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use serde::Deserialize;
 use serde_json::Value;
+use std::collections::HashMap;
 use std::fmt::Debug;
 use std::time::Duration;
 use tedge_mqtt_ext::Topic;
 use tedge_mqtt_ext::TopicFilter;
+use tokio::fs::read_dir;
+use tokio::fs::read_to_string;
+use tracing::error;
+use tracing::info;
 
 #[derive(Deserialize)]
 pub struct FlowConfig {
@@ -101,6 +106,53 @@ pub enum ConfigError {
 }
 
 impl FlowConfig {
+    pub async fn load_all_flows(config_dir: &Utf8Path) -> HashMap<Utf8PathBuf, FlowConfig> {
+        let mut flows = HashMap::new();
+        let Ok(mut entries) = read_dir(config_dir).await.map_err(
+            |err| error!(target: "flows", "Failed to read flows from {config_dir}: {err}"),
+        ) else {
+            return flows;
+        };
+
+        while let Ok(Some(entry)) = entries.next_entry().await {
+            let Some(path) = Utf8Path::from_path(&entry.path()).map(|p| p.to_path_buf()) else {
+                error!(target: "flows", "Skipping non UTF8 path: {}", entry.path().display());
+                continue;
+            };
+            if let Ok(file_type) = entry.file_type().await {
+                if file_type.is_file() {
+                    if let Some("toml") = path.extension() {
+                        info!(target: "flows", "Loading flow: {path}");
+                        if let Some(flow) = FlowConfig::load_single_flow(&path).await {
+                            flows.insert(path.clone(), flow);
+                        }
+                    }
+                }
+            }
+        }
+        flows
+    }
+
+    pub async fn load_single_flow(flow: &Utf8Path) -> Option<FlowConfig> {
+        match FlowConfig::load_flow(flow).await {
+            Ok(flow) => Some(flow),
+            Err(err) => {
+                error!(target: "flows", "Failed to load flow {flow}: {err}");
+                None
+            }
+        }
+    }
+
+    pub fn wrap_script_into_flow(script: &Utf8Path) -> FlowConfig {
+        FlowConfig::from_step(script.to_owned())
+    }
+
+    async fn load_flow(path: &Utf8Path) -> Result<FlowConfig, LoadError> {
+        let specs = read_to_string(path).await?;
+        let flow: FlowConfig = toml::from_str(&specs)?;
+        Ok(flow)
+    }
+
     pub fn from_step(script: Utf8PathBuf) -> Self {
         let input_topic = "#".to_string();
         let step = StepConfig {

crates/extensions/tedge_flows/src/flow.rs

@@ -117,6 +117,18 @@ pub enum FlowError {
     Anyhow(#[from] anyhow::Error),
 }
 
+impl AsRef<Flow> for Flow {
+    fn as_ref(&self) -> &Flow {
+        self
+    }
+}
+
+impl AsMut<Flow> for Flow {
+    fn as_mut(&mut self) -> &mut Flow {
+        self
+    }
+}
+
 impl Flow {
     pub fn name(&self) -> &str {
         self.source.as_str()
@@ -176,7 +188,7 @@ impl Flow {
         Ok(messages)
     }
 
-    pub fn accept_message(&mut self, source: &SourceTag, message: &Message) -> bool {
+    pub fn accept_message(&self, source: &SourceTag, message: &Message) -> bool {
         self.input.accept_message(source, message)
     }
 

crates/extensions/tedge_flows/src/lib.rs

@@ -1,6 +1,6 @@
 mod actor;
 mod config;
-pub mod flow;
+mod flow;
 mod input_source;
 mod js_lib;
 mod js_runtime;
@@ -12,6 +12,9 @@ mod stats;
 
 use crate::actor::FlowsMapper;
 use crate::actor::STATS_DUMP_INTERVAL;
+pub use crate::flow::*;
+pub use crate::registry::BaseFlowRegistry;
+pub use crate::registry::FlowRegistryExt;
 pub use crate::runtime::MessageProcessor;
 use camino::Utf8Path;
 use std::collections::HashSet;
@@ -38,6 +41,7 @@ use tedge_watch_ext::WatchEvent;
 use tedge_watch_ext::WatchRequest;
 use tokio::time::Instant;
 use tracing::error;
+
 fan_in_message_type!(InputMessage[MqttMessage, WatchEvent, FsWatchEvent, Tick]: Clone, Debug, Eq, PartialEq);
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -47,12 +51,13 @@ pub struct FlowsMapperBuilder {
     message_box: SimpleMessageBoxBuilder<InputMessage, SubscriptionDiff>,
     mqtt_sender: DynSender<MqttMessage>,
     watch_request_sender: DynSender<WatchRequest>,
-    processor: MessageProcessor,
+    processor: MessageProcessor<BaseFlowRegistry>,
 }
 
 impl FlowsMapperBuilder {
     pub async fn try_new(config_dir: impl AsRef<Utf8Path>) -> Result<Self, LoadError> {
-        let mut processor = MessageProcessor::try_new(config_dir).await?;
+        let registry = BaseFlowRegistry::new(config_dir);
+        let mut processor = MessageProcessor::try_new(registry).await?;
         let message_box = SimpleMessageBoxBuilder::new("TedgeFlows", 16);
         let mqtt_sender = NullSender.into();
         let watch_request_sender = NullSender.into();