[DataBroker Rewrite] Sim MVP

.config/nextest.toml

@@ -0,0 +1,7 @@
+[profile.ci]
+failure-output = "immediate-final"
+
+[profile.ci.junit]  # this can be some other profile, too
+path = "junit.xml"
+store-success-output = true
+store-failure-output = true

.github/workflows/rust.yaml

@@ -2,10 +2,9 @@ name: Rust
 
 on:
   push:
-    branches: ["main"]
+    branches:
+      - main
   pull_request:
-    branches: ["main"]
-
 env:
   CARGO_TERM_COLOR: always
 
@@ -29,33 +28,33 @@ jobs:
       - uses: taiki-e/install-action@nextest
       - uses: actions/checkout@v4
       - name: Install Rust
-        run: rustup update nightly 
+        run: |
+          rustup update nightly 
       - name: Install cargo-llvm-cov
         run: rustup component add llvm-tools-preview --toolchain nightly
       - uses: taiki-e/install-action@cargo-llvm-cov
       - uses: taiki-e/install-action@nextest
       - name: Collect coverage data (including doctests)
         run: |
           cargo llvm-cov --no-report nextest --ignore-filename-regex calloop-0.10.6/*
-          cargo llvm-cov --no-report --doc  --ignore-filename-regex calloop-0.10.6/*
-          cargo llvm-cov report --doctests --lcov --output-path lcov.info  --ignore-filename-regex calloop-0.10.6/*
+          cargo llvm-cov report --lcov --output-path lcov.info  --ignore-filename-regex calloop-0.10.6/*
       - name: Upload coverage
         uses: codecov/codecov-action@v1
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
-          fail_ci_if_error: true
-  docs:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - uses: taiki-e/install-action@nextest
-      - uses: actions/checkout@v4
-      - name: Install Rust
-        run: rustup update stable
-      - name: Build docs
-        run: cargo doc --verbose --all-features --workspace
-      - name: Deploy docs
-        uses: peaceiris/actions-gh-pages@v3
-        with:
-            github_token: ${{ secrets.GITHUB_TOKEN }}
-            publish_dir: ./target/doc
+          fail_ci_if_error: false
+#  docs:
+#    runs-on: ubuntu-latest
+#    steps:
+#      - uses: actions/checkout@v3
+#      - uses: taiki-e/install-action@nextest
+#      - uses: actions/checkout@v4
+#      - name: Install Rust
+#        run: rustup update stable
+#      - name: Build docs
+#        run: cargo doc --verbose --all-features --workspace
+#      - name: Deploy docs
+#        uses: peaceiris/actions-gh-pages@v3
+#        with:
+#            github_token: ${{ secrets.GITHUB_TOKEN }}
+#            publish_dir: ./target/doc

.vscode/settings.json

@@ -1,5 +1,8 @@
 {
     "rust-analyzer.linkedProjects": [
         "./lil-broker/Cargo.toml"
+    ],
+    "cSpell.words": [
+        "tasker"
     ]
 }

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 
 members = [
-    "lil-broker", "lil-helper", "lil-quad", "lil-sym",
+    "lil-broker", "lil-helper", "lil-quad", "lil-sym", "lil-viz",
 ]
 resolver = "2"

lil-broker/Cargo.toml

@@ -14,3 +14,5 @@ serde = { version = "1.0.197", features = ["derive"] }
 serde_json = "1.0.117"
 tracing = {version = "0.1.40", features = ["log"]}
 json-unflattening = "0.1.2"
+lil-helper = { path = "../lil-helper" }
+anyhow = "1.0.82"

lil-broker/src/buckets/mod.rs

@@ -3,7 +3,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use tracing::debug;
 
 mod querying;
-
+#[derive(Debug)]
 pub struct Bucket {
     pub name: String,
     pub bucket_tags: BTreeSet<Tag>,

lil-broker/src/buckets/querying.rs

@@ -81,7 +81,7 @@ impl Bucket {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{types::Tag, Primatives, Timestamp};
+    use crate::{Primatives, Timestamp};
 
     /// Querying tests
     #[test]

lil-broker/src/database/mod.rs

@@ -1,20 +1,27 @@
 mod query;
 pub use query::*;
 use std::collections::BTreeMap;
-use tracing::{debug, error, info};
-
-use crate::{Bucket, Tag};
+use tracing::{error, info};
 
+use crate::{Bucket, Primatives, Tag, Timestamp};
+#[derive(Debug)]
 pub struct Database {
+    pub current_t: Option<Timestamp>,
     pub buckets: BTreeMap<String, Bucket>,
 }
 
 impl Database {
     pub fn new() -> Database {
         Database {
             buckets: BTreeMap::new(),
+            current_t: None,
         }
     }
+
+    pub fn set_time(&mut self, time: Timestamp) {
+        self.current_t = Some(time);
+    }
+
     pub fn get_keys(&self) -> Vec<String> {
         self.buckets.keys().cloned().collect()
     }
@@ -27,6 +34,11 @@ impl Database {
         }
     }
 
+    pub fn quick_write(&mut self, topic: &str, data: Primatives) -> Result<QueryResponse, String> {
+        let query = WriteQuery::new(topic.to_string(), data, self.current_t.unwrap());
+        self.query_write(query)
+    }
+
     pub fn add_tag_to_bucket(&mut self, bucket_name: &str, tag: Tag) {
         //Create a new bucket if it doesn't exist
         if !self.buckets.contains_key(bucket_name) {

lil-broker/src/database/query/get_latest.rs

@@ -1,7 +1,7 @@
 use crate::{Database, QueryCommand, QueryResponse};
 
 use super::tag_filter::TagFilter;
-use tracing::{debug, info};
+use tracing::debug;
 #[derive(Debug, Clone)]
 pub struct GetLatestQuery {
     pub topics: Vec<String>,
@@ -119,14 +119,42 @@
 
         Ok(response)
     }
+
+    pub fn query_get_latest_stripped(
+        &mut self,
+        query: GetLatestQuery,
+    ) -> Result<QueryResponse, String> {
+        let mut response = QueryResponse::default();
+        debug!("Querying for latest data: {:?}", query);
+        let all_bucket_keys = self.get_keys().into_iter();
+        let matching_keys = all_bucket_keys.filter(|key| {
+            for topic in &query.topics {
+                if key.starts_with(topic) {
+                    return true;
+                }
+            }
+            false
+        });
+        // Read the matching keys
+        let addtional_keys = self.read_bucket(matching_keys.collect(), &query, &mut response);
+
+        if let Some(addtional_keys) = addtional_keys {
+            debug!("Additional keys to read: {:?}", addtional_keys);
+            self.read_bucket(addtional_keys, &query, &mut response);
+        }
+
+        Ok(response)
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use crate::{Primatives, Tag, Timestamp, WriteQuery};
-    use pretty_assertions::{assert_eq, assert_ne};
+    use crate::{Primatives, Timestamp, WriteQuery};
+    use pretty_assertions::assert_eq;
     use serde_json::json;
 
+    use tracing::info;
+
     use super::*;
 
     #[test]
@@ -154,7 +182,6 @@
 
     #[test]
     fn test_get_latest_json_struct() {
-        env_logger::init();
         let mut db = Database::new();
 
         #[derive(Debug, Clone, PartialEq, serde::Deserialize, serde::Serialize)]
@@ -239,7 +266,6 @@
 
     #[test]
     fn test_get_latest_query_ack() {
-        env_logger::init();
         let mut db = Database::new();
         let query1 = WriteQuery::new("test/a/1".into(), 1.0.into(), Timestamp::from_seconds(1.0));
         let _write_res = db.query_batch(vec![query1.into()]).unwrap();
@@ -271,7 +297,6 @@
 
     #[test]
     fn test_get_latest_query_bucket_tags() {
-        env_logger::init();
         let mut db = Database::new();
         let query1 = WriteQuery::new("test/a/1".into(), 1.0.into(), Timestamp::from_seconds(1.0));
         db.add_tag_to_bucket("test/a/1", "user/test_tag".into());
@@ -293,7 +318,6 @@
 
     #[test]
     fn test_get_latest_query_additonial() {
-        env_logger::init();
         let mut db = Database::new();
 
         let queries = vec![

lil-broker/src/database/query/lookup_range.rs

@@ -99,7 +99,7 @@ mod tests {
     use serde_json::json;
     use tracing::info;
 
-    use crate::{Primatives, Tag, Timestamp, WriteQuery};
+    use crate::{Timestamp, WriteQuery};
 
     use super::*;
 
@@ -129,7 +129,6 @@ mod tests {
     }
     #[test]
     fn test_lookup_range_basic() {
-        env_logger::init();
         let mut db = generate_data();
 
         let read_query = LookupRangeQuery {
@@ -149,7 +148,6 @@ mod tests {
 
     #[test]
     fn test_get_latest_json_struct() {
-        env_logger::init();
         let mut db = Database::new();
 
         #[derive(Debug, Clone, PartialEq, serde::Deserialize, serde::Serialize)]
@@ -213,7 +211,6 @@ mod tests {
 
     #[test]
     fn test_lookup_range_wild_card() {
-        env_logger::init();
         let mut db = generate_data();
 
         let read_query = LookupRangeQuery {

lil-broker/src/database/query/mod.rs

@@ -1,6 +1,6 @@
 use std::collections::BTreeMap;
 
-use crate::DataPoint;
+use crate::{DataPoint, Primatives, Timestamp};
 
 mod get_latest;
 mod lookup;
@@ -11,10 +11,13 @@ mod write;
 pub use get_latest::*;
 use json_unflattening::unflattening::unflatten;
 pub use lookup::*;
+
 pub use lookup_range::*;
+
 use serde_json::{json, Value};
 pub use tag_filter::*;
 
+use tracing::debug;
 pub use write::*;
 #[derive(Debug, Clone)]
 pub enum QueryCommand {
@@ -45,6 +48,31 @@ impl QueryResponse {
         }
     }
 
+    pub fn from_data(data: BTreeMap<String, Vec<DataPoint>>) -> QueryResponse {
+        let len = data.len();
+        QueryResponse {
+            data,
+            metadata: QueryResponseMetadata {
+                n_results: len,
+                was_successful: true,
+            },
+        }
+    }
+
+    pub fn from_json(json: Value) -> QueryResponse {
+        let mut data = BTreeMap::new();
+        for (key, value) in json.as_object().unwrap() {
+            let mut data_points = Vec::new();
+            for (timestamp, data) in value.as_object().unwrap() {
+                let timestamp = Timestamp::new(timestamp.parse().unwrap());
+                let data = Primatives::from_value(data.clone()).unwrap();
+                data_points.push(DataPoint::new(timestamp, data));
+            }
+            data.insert(key.to_string(), data_points);
+        }
+        QueryResponse::from_data(data)
+    }
+
     pub fn to_json(&self, prefix: &str) -> Value {
         let mut values: serde_json::Map<String, Value> = serde_json::Map::new();
         for (key, data) in self.data.iter() {
@@ -57,8 +85,9 @@ impl QueryResponse {
             values.insert(key.to_string(), json!(data_points.last()));
         }
         // Strip Prefix
+        debug!("{:?}", values);
 
-        let unflattened_json = unflatten(&values).unwrap();
+        let unflattened_json = unflatten(&values).unwrap_or_default();
         unflattened_json
     }
 

lil-broker/src/database/query/write.rs

@@ -1,10 +1,10 @@
 use crate::{Bucket, DataPoint, Database, Primatives, QueryCommand, QueryResponse, Tag, Timestamp};
 
-use flatten_json_object::{ArrayFormatting, Flattener};
 use json_unflattening::flattening::flatten;
-use serde::de::value;
+
 use serde_json::Value;
-use tracing::{error, info};
+use tracing::info;
+
 #[derive(Debug, Clone)]
 pub struct WriteQuery {
     pub topic: String,
@@ -120,7 +120,7 @@ mod tests {
 
     #[test]
     fn test_write_query_from_json_complex() {
-        env_logger::init();
+        //env_logger::init();
 
         let json = json!(
         {