Refactor database into separate module, and add in memory database for testing

Author: jarhodes314

Cargo.lock

@@ -75,7 +75,7 @@ impl TestCommand {
             }
         };
         processor
-            .on_message(MessageSource::MQTT, timestamp, &message)
+            .on_message(MessageSource::Mqtt, timestamp, &message)
             .await
             .into_iter()
             .for_each(|msg| self.print_messages(msg))

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

@@ -35,6 +35,8 @@ toml = { workspace = true, features = ["parse"] }
 tracing = { workspace = true }
 
 [dev-dependencies]
+env_logger = { workspace = true }
+rstest = { workspace = true }
 tedge_mqtt_ext = { workspace = true, features = ["test-helpers"] }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["test-util"] }

crates/extensions/tedge_flows/Cargo.toml

@@ -52,7 +52,7 @@ impl Actor for FlowsMapper {
                 }
                 InputMessage::MqttMessage(message) => match Message::try_from(message) {
                     Ok(message) => {
-                        self.on_message(MessageSource::MQTT, DateTime::now(), message)
+                        self.on_message(MessageSource::Mqtt, DateTime::now(), message)
                             .await?
                     }
                     Err(err) => {
@@ -142,7 +142,7 @@ impl FlowsMapper {
             match flow_messages {
                 Ok(messages) => self.publish_messages(flow_id, timestamp, messages).await?,
                 Err(err) => {
-                    error!(target: "flows", "{flow_id}: {err}");
+                    error!(target: "flows", "{flow_id}: {err:#}");
                 }
             }
         }
@@ -192,7 +192,7 @@ impl FlowsMapper {
     ) -> Result<(), RuntimeError> {
         if let Some(flow) = self.processor.flows.get(&flow_id) {
             match &flow.output {
-                FlowOutput::MQTT { output_topics } => {
+                FlowOutput::Mqtt { output_topics } => {
                     for message in messages {
                         match MqttMessage::try_from(message) {
                             Ok(message) if output_topics.accept_topic(&message.topic) => {
@@ -212,6 +212,7 @@ impl FlowsMapper {
                 FlowOutput::MeaDB { output_series } => {
                     for message in messages {
                         info!(target: "flows", "store {output_series} @{}.{} [{}]", timestamp.seconds, timestamp.nanoseconds, message.topic);
+                        let timestamp = DateTime::now();
                         if let Err(err) = self
                             .processor
                             .database

crates/extensions/tedge_flows/src/actor.rs

@@ -1,4 +1,6 @@
 use anyhow::Context;
+use tedge_flows::database::FjallMeaDb;
+use tedge_flows::database::MeaDb;
 use tedge_flows::flow::DateTime;
 use tedge_flows::flow::Message;
 
@@ -12,7 +14,7 @@ async fn main() -> anyhow::Result<()> {
     let series_name = args.nth(1).unwrap_or("latest-data-points".to_string());
     println!("Reading series name: {series_name}");
 
-    let mut db = tedge_flows::MeaDB::open(DB_PATH)
+    let mut db = FjallMeaDb::open(DB_PATH)
         .await
         .with_context(|| format!("Failed to open DB at path={DB_PATH}"))?;
 

crates/extensions/tedge_flows/src/bin/db_dump.rs

@@ -163,7 +163,7 @@ impl TryFrom<InputConfig> for FlowInput {
 
     fn try_from(input: InputConfig) -> Result<Self, Self::Error> {
         match input {
-            InputConfig::Mqtt { topics } => Ok(FlowInput::MQTT {
+            InputConfig::Mqtt { topics } => Ok(FlowInput::Mqtt {
                 topics: topic_filters(topics)?,
             }),
             InputConfig::MeaDB {
@@ -184,7 +184,7 @@ impl TryFrom<OutputConfig> for FlowOutput {
 
     fn try_from(output: OutputConfig) -> Result<Self, Self::Error> {
         match output {
-            OutputConfig::Mqtt { topics } => Ok(FlowOutput::MQTT {
+            OutputConfig::Mqtt { topics } => Ok(FlowOutput::Mqtt {
                 output_topics: topic_filters(topics)?,
             }),
             OutputConfig::MeaDB { series } => Ok(FlowOutput::MeaDB {