Separate Flow vs ConnectedFlow

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

Author: didier-wenzek

crates/extensions/tedge_flows/src/actor.rs

@@ -1,10 +1,10 @@
+use crate::connected_flow::ConnectedFlowRegistry;
 use crate::flow::DateTime;
 use crate::flow::FlowError;
 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;
@@ -46,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<BaseFlowRegistry>,
+    pub(super) processor: MessageProcessor<ConnectedFlowRegistry>,
     pub(super) next_dump: Instant,
 }
 
@@ -71,7 +71,7 @@ impl Actor for FlowsMapper {
                     self.on_message(source, Message::from(message)).await?
                 }
                 InputMessage::WatchEvent(event) => {
-                    self.on_process_event(event).await?;
+                    self.on_input_event(event).await?;
                 }
                 InputMessage::FsWatchEvent(FsWatchEvent::Modified(path)) => {
                     let Ok(path) = Utf8PathBuf::try_from(path) else {
@@ -166,7 +166,7 @@ impl FlowsMapper {
             new_watched_commands.insert(topic.to_owned());
         }
         for old_command in self.watched_commands.drain() {
-            info!(target: "flows", "removing input: {}", old_command);
+            info!(target: "flows", "Removing input: {}", old_command);
             watch_requests.push(WatchRequest::UnWatch { topic: old_command });
         }
         self.watched_commands = new_watched_commands;
@@ -208,8 +208,29 @@ impl FlowsMapper {
     async fn on_source_poll(&mut self) -> Result<(), RuntimeError> {
         let now = Instant::now();
         let timestamp = DateTime::now();
-        for messages in self.processor.on_source_poll(timestamp, now).await {
-            self.publish_result(messages).await?;
+
+        let mut in_messages = vec![];
+        for flow in self.processor.registry.flows_mut() {
+            in_messages.push(flow.on_source_poll(timestamp, now).await);
+        }
+
+        for messages in in_messages {
+            match messages {
+                FlowResult::Ok { flow, messages, .. } => {
+                    for message in messages {
+                        if let Some(flow_output) = self
+                            .processor
+                            .on_flow_input(&flow, timestamp, &message)
+                            .await
+                        {
+                            self.publish_result(flow_output).await?;
+                        }
+                    }
+                }
+                poll_error => {
+                    self.publish_result(poll_error).await?;
+                }
+            }
         }
 
         Ok(())
@@ -247,43 +268,48 @@ impl FlowsMapper {
         Ok(())
     }
 
-    async fn on_process_event(&mut self, event: WatchEvent) -> Result<(), RuntimeError> {
+    async fn on_input_event(&mut self, event: WatchEvent) -> Result<(), RuntimeError> {
         match event {
             WatchEvent::StdoutLine { topic, line } => {
-                self.on_process_message(topic, line).await?;
+                self.on_input_message(topic, line).await?;
             }
             WatchEvent::StderrLine { topic, line } => {
                 warn!(target: "flows", "Input command {topic}: {line}");
             }
             WatchEvent::Error { topic, error } => {
                 error!(target: "flows", "Cannot monitor command: {error}");
-                self.on_process_error(&topic, error.into()).await?;
+                self.on_input_error(&topic, error.into()).await?;
             }
             WatchEvent::EndOfStream { topic } => {
                 error!(target: "flows", "End of input stream: {topic}");
-                self.on_process_eos(&topic).await?
+                self.on_input_eos(&topic).await?
             }
         }
         Ok(())
     }
 
-    async fn on_process_message(
+    async fn on_input_message(
         &mut self,
         flow_name: String,
         line: String,
     ) -> Result<(), RuntimeError> {
         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(),
-            };
-            self.on_message(source, Message::new(topic, line)).await?;
+            let topic = flow.input_topic().to_string();
+            let timestamp = DateTime::now();
+            let message = Message::new(topic, line);
+            if let Some(result) = self
+                .processor
+                .on_flow_input(&flow_name, timestamp, &message)
+                .await
+            {
+                self.publish_result(result).await?;
+            }
         }
 
         Ok(())
     }
 
-    async fn on_process_error(
+    async fn on_input_error(
         &mut self,
         flow_name: &str,
         error: FlowError,
@@ -316,7 +342,7 @@ impl FlowsMapper {
         Ok(())
     }
 
-    async fn on_process_eos(&mut self, flow_name: &str) -> Result<(), RuntimeError> {
+    async fn on_input_eos(&mut self, flow_name: &str) -> Result<(), RuntimeError> {
         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);

crates/extensions/tedge_flows/src/config.rs

@@ -1,10 +1,7 @@
 use crate::flow::Flow;
+use crate::flow::FlowInput;
 use crate::flow::FlowOutput;
 use crate::flow::FlowStep;
-use crate::input_source::CommandFlowInput;
-use crate::input_source::FileFlowInput;
-use crate::input_source::FlowInput;
-use crate::input_source::MqttFlowInput;
 use crate::js_runtime::JsRuntime;
 use crate::js_script::JsScript;
 use crate::LoadError;
@@ -177,8 +174,7 @@ impl FlowConfig {
         config_dir: &Utf8Path,
         source: Utf8PathBuf,
     ) -> Result<Flow, ConfigError> {
-        let flow_name = source.clone().to_string();
-        let input = self.input.compile(flow_name)?;
+        let input = self.input.try_into()?;
         let output = self.output.try_into()?;
         let errors = self.errors.try_into()?;
         let mut steps = vec![];
@@ -223,27 +219,42 @@ impl StepConfig {
     }
 }
 
-impl InputConfig {
-    pub fn compile(self, flow_name: String) -> Result<Box<dyn FlowInput>, ConfigError> {
-        Ok(match self {
-            InputConfig::Mqtt { topics } => Box::new(MqttFlowInput {
+impl TryFrom<InputConfig> for FlowInput {
+    type Error = ConfigError;
+    fn try_from(input: InputConfig) -> Result<Self, Self::Error> {
+        Ok(match input {
+            InputConfig::Mqtt { topics } => FlowInput::Mqtt {
                 topics: topic_filters(topics)?,
-            }),
+            },
             InputConfig::File {
                 topic,
                 path,
                 interval,
             } => {
                 let topic = topic.unwrap_or_else(|| path.clone().to_string());
-                Box::new(FileFlowInput::new(flow_name, topic, path, interval))
+                match interval {
+                    Some(interval) if !interval.is_zero() => FlowInput::PollFile {
+                        topic,
+                        path,
+                        interval,
+                    },
+                    _ => FlowInput::StreamFile { topic, path },
+                }
             }
             InputConfig::Process {
                 topic,
                 command,
                 interval,
             } => {
                 let topic = topic.unwrap_or_else(|| command.clone());
-                Box::new(CommandFlowInput::new(flow_name, topic, command, interval))
+                match interval {
+                    Some(interval) if !interval.is_zero() => FlowInput::PollCommand {
+                        topic,
+                        command,
+                        interval,
+                    },
+                    _ => FlowInput::StreamCommand { topic, command },
+                }
             }
         })
     }

crates/extensions/tedge_flows/src/connected_flow.rs

@@ -0,0 +1,149 @@
+use crate::config::ConfigError;
+use crate::flow::DateTime;
+use crate::flow::Flow;
+use crate::flow::FlowError;
+use crate::flow::FlowInput;
+use crate::flow::FlowResult;
+use crate::flow::Message;
+use crate::input_source::CommandFlowInput;
+use crate::input_source::FileFlowInput;
+use crate::input_source::FlowSource;
+use crate::input_source::MqttFlowInput;
+use crate::registry::FlowRegistry;
+use crate::registry::FlowStore;
+use camino::Utf8Path;
+use tedge_watch_ext::WatchRequest;
+use tokio::time::Instant;
+
+/// A flow connected to a source of messages
+pub struct ConnectedFlow {
+    flow: Flow,
+    pub(crate) input: Box<dyn FlowSource>,
+}
+
+impl AsRef<Flow> for ConnectedFlow {
+    fn as_ref(&self) -> &Flow {
+        &self.flow
+    }
+}
+
+impl AsMut<Flow> for ConnectedFlow {
+    fn as_mut(&mut self) -> &mut Flow {
+        &mut self.flow
+    }
+}
+
+impl ConnectedFlow {
+    pub fn new(flow: Flow) -> Self {
+        let name = flow.name().to_string();
+        let input = connect(name, flow.input.clone());
+        ConnectedFlow { flow, input }
+    }
+
+    pub fn name(&self) -> &str {
+        self.flow.name()
+    }
+
+    pub fn input_topic(&self) -> &str {
+        self.input.enforced_topic().unwrap_or_default()
+    }
+
+    pub fn watch_request(&self) -> Option<WatchRequest> {
+        self.input.watch_request()
+    }
+
+    pub async fn on_source_poll(&mut self, timestamp: DateTime, now: Instant) -> FlowResult {
+        let result = self.on_source_poll_steps(timestamp, now).await;
+        self.flow.publish(result)
+    }
+
+    async fn on_source_poll_steps(
+        &mut self,
+        timestamp: DateTime,
+        now: Instant,
+    ) -> Result<Vec<Message>, FlowError> {
+        let source = &mut self.input;
+        if !source.is_ready(now) {
+            return Ok(vec![]);
+        };
+
+        let messages = source.poll(timestamp).await?;
+        source.update_after_poll(now);
+        Ok(messages)
+    }
+
+    pub fn on_error(&self, error: FlowError) -> FlowResult {
+        self.flow.publish(Err(error))
+    }
+}
+
+fn connect(flow_name: String, input: FlowInput) -> Box<dyn FlowSource> {
+    match input {
+        FlowInput::Mqtt { topics } => Box::new(MqttFlowInput { topics }),
+        FlowInput::PollFile {
+            topic,
+            path,
+            interval,
+        } => Box::new(FileFlowInput::new(flow_name, topic, path, Some(interval))),
+        FlowInput::PollCommand {
+            topic,
+            command,
+            interval,
+        } => Box::new(CommandFlowInput::new(
+            flow_name,
+            topic,
+            command,
+            Some(interval),
+        )),
+        FlowInput::StreamFile { topic, path } => {
+            Box::new(FileFlowInput::new(flow_name, topic, path, None))
+        }
+        FlowInput::StreamCommand { topic, command } => {
+            Box::new(CommandFlowInput::new(flow_name, topic, command, None))
+        }
+    }
+}
+
+pub struct ConnectedFlowRegistry {
+    flows: FlowStore<ConnectedFlow>,
+}
+
+impl ConnectedFlowRegistry {
+    pub fn new(config_dir: impl AsRef<Utf8Path>) -> Self {
+        ConnectedFlowRegistry {
+            flows: FlowStore::new(config_dir),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl FlowRegistry for ConnectedFlowRegistry {
+    type Flow = ConnectedFlow;
+
+    fn compile(flow: Flow) -> Result<Self::Flow, ConfigError> {
+        Ok(ConnectedFlow::new(flow))
+    }
+
+    fn store(&self) -> &FlowStore<Self::Flow> {
+        &self.flows
+    }
+
+    fn store_mut(&mut self) -> &mut FlowStore<Self::Flow> {
+        &mut self.flows
+    }
+
+    fn deadlines(&self) -> impl Iterator<Item = Instant> + '_ {
+        let script_deadlines = self
+            .flows
+            .flows()
+            .flat_map(|flow| &flow.as_ref().steps)
+            .filter_map(|step| step.script.next_execution);
+
+        let source_deadlines = self
+            .flows
+            .flows()
+            .filter_map(|flow| flow.input.next_deadline());
+
+        script_deadlines.chain(source_deadlines)
+    }
+}