Separate streaming sources from polling ones

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

Author: didier-wenzek

crates/extensions/tedge_flows/src/actor.rs

@@ -159,7 +159,7 @@ impl FlowsMapper {
                 continue;
             };
             if !self.watched_commands.contains(topic) {
-                info!(target: "flows", "Adding input: {}", flow.input);
+                info!(target: "flows", "Adding input: {}", flow.as_ref().input);
                 watch_requests.push(request);
             }
             self.watched_commands.remove(topic);
@@ -316,7 +316,7 @@ impl FlowsMapper {
     ) -> Result<(), RuntimeError> {
         let Some((info, flow_error)) = self.processor.registry.flow(flow_name).map(|flow| {
             (
-                format!("Reconnecting input: {flow_name}: {}", flow.input),
+                format!("Reconnecting input: {flow_name}: {}", flow.as_ref().input),
                 flow.on_error(error),
             )
         }) else {
@@ -345,7 +345,7 @@ impl FlowsMapper {
     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);
+                info!(target: "flows", "Reconnecting input: {flow_name}: {}", flow.as_ref().input);
                 self.watch_request_sender.send(request).await?
             };
         }

crates/extensions/tedge_flows/src/connected_flow.rs

@@ -5,10 +5,12 @@ 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::input_source::CommandPollingSource;
+use crate::input_source::CommandStreamingSource;
+use crate::input_source::FilePollingSource;
+use crate::input_source::FileStreamingSource;
+use crate::input_source::PollingSource;
+use crate::input_source::StreamingSource;
 use crate::registry::FlowRegistry;
 use crate::registry::FlowStore;
 use camino::Utf8Path;
@@ -18,7 +20,8 @@ use tokio::time::Instant;
 /// A flow connected to a source of messages
 pub struct ConnectedFlow {
     flow: Flow,
-    pub(crate) input: Box<dyn FlowSource>,
+    streaming_source: Option<Box<dyn StreamingSource>>,
+    polling_source: Option<Box<dyn PollingSource>>,
 }
 
 impl AsRef<Flow> for ConnectedFlow {
@@ -35,21 +38,31 @@ impl AsMut<Flow> for ConnectedFlow {
 
 impl ConnectedFlow {
     pub fn new(flow: Flow) -> Self {
-        let name = flow.name().to_string();
-        let input = connect(name, flow.input.clone());
-        ConnectedFlow { flow, input }
+        let streaming_source = streaming_source(flow.name().to_owned(), flow.input.clone());
+        let polling_source = polling_source(flow.input.clone());
+        ConnectedFlow {
+            flow,
+            streaming_source,
+            polling_source,
+        }
     }
 
     pub fn name(&self) -> &str {
         self.flow.name()
     }
 
     pub fn input_topic(&self) -> &str {
-        self.input.enforced_topic().unwrap_or_default()
+        self.flow.input.enforced_topic().unwrap_or_default()
     }
 
     pub fn watch_request(&self) -> Option<WatchRequest> {
-        self.input.watch_request()
+        self.streaming_source
+            .as_ref()
+            .and_then(|source| source.watch_request())
+    }
+
+    pub fn next_deadline(&self) -> Option<Instant> {
+        self.polling_source.as_ref().and_then(|p| p.next_deadline())
     }
 
     pub async fn on_source_poll(&mut self, timestamp: DateTime, now: Instant) -> FlowResult {
@@ -62,7 +75,9 @@ impl ConnectedFlow {
         timestamp: DateTime,
         now: Instant,
     ) -> Result<Vec<Message>, FlowError> {
-        let source = &mut self.input;
+        let Some(source) = &mut self.polling_source.as_mut() else {
+            return Ok(vec![]);
+        };
         if !source.is_ready(now) {
             return Ok(vec![]);
         };
@@ -77,30 +92,37 @@ impl ConnectedFlow {
     }
 }
 
-fn connect(flow_name: String, input: FlowInput) -> Box<dyn FlowSource> {
+fn streaming_source(flow_name: String, input: FlowInput) -> Option<Box<dyn StreamingSource>> {
+    match input {
+        FlowInput::StreamFile { topic: _, path } => {
+            Some(Box::new(FileStreamingSource::new(flow_name, path)))
+        }
+
+        FlowInput::StreamCommand { topic: _, command } => {
+            Some(Box::new(CommandStreamingSource::new(flow_name, command)))
+        }
+
+        _ => None,
+    }
+}
+
+fn polling_source(input: FlowInput) -> Option<Box<dyn PollingSource>> {
     match input {
-        FlowInput::Mqtt { topics } => Box::new(MqttFlowInput { topics }),
         FlowInput::PollFile {
             topic,
             path,
             interval,
-        } => Box::new(FileFlowInput::new(flow_name, topic, path, Some(interval))),
+        } => Some(Box::new(FilePollingSource::new(topic, path, 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))
-        }
+        } => Some(Box::new(CommandPollingSource::new(
+            topic, command, interval,
+        ))),
+
+        _ => None,
     }
 }
 
@@ -139,10 +161,7 @@ impl FlowRegistry for ConnectedFlowRegistry {
             .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());
+        let source_deadlines = self.flows.flows().filter_map(|flow| flow.next_deadline());
 
         script_deadlines.chain(source_deadlines)
     }

crates/extensions/tedge_flows/src/flow.rs

@@ -7,6 +7,8 @@ use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use serde_json::json;
 use serde_json::Value;
+use std::fmt::Display;
+use std::fmt::Formatter;
 use std::time::Duration;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
@@ -323,6 +325,28 @@ impl Flow {
     }
 }
 
+impl Display for FlowInput {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            FlowInput::Mqtt { topics } => {
+                write!(f, "MQTT topics: {:?}", topics)
+            }
+            FlowInput::PollFile { path, .. } => {
+                write!(f, "Polling file: {path}")
+            }
+            FlowInput::PollCommand { command, .. } => {
+                write!(f, "Polling command: {command}")
+            }
+            FlowInput::StreamFile { path, .. } => {
+                write!(f, "Streaming file: {path}")
+            }
+            FlowInput::StreamCommand { command, .. } => {
+                write!(f, "Streaming command: {command}")
+            }
+        }
+    }
+}
+
 impl FlowInput {
     pub fn topics(&self) -> TopicFilter {
         match self {
@@ -331,6 +355,16 @@ impl FlowInput {
         }
     }
 
+    pub fn enforced_topic(&self) -> Option<&str> {
+        match self {
+            FlowInput::Mqtt { .. } => None,
+            FlowInput::PollFile { topic, .. }
+            | FlowInput::PollCommand { topic, .. }
+            | FlowInput::StreamFile { topic, .. }
+            | FlowInput::StreamCommand { topic, .. } => Some(topic),
+        }
+    }
+
     pub fn accept_message(&self, message: &Message) -> bool {
         match self {
             FlowInput::Mqtt { topics } => topics.accept_topic_name(&message.topic),