-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: on-device bus with
rumqttd and data joiner on uplink (#348)
* doc: example config * feat: config types to deserialize example * feat: service bus traits * feat: on device bus and data joiner * feat: simulator as a service on bus * style: fmt toml * remove dbg * fix: spawn blocking thread task * fix: routing setup * refactor: pass whole json * feat: `select_ fields = "all"` * feat: instant data push * fix: deserialization error message * fix: actually push data instantly * fix: handle timestamp and sequence from incoming stream * remove dbg * ci: clippy suggestion * Bus at same level as TcpJson * test: data and status * Make structs PartialEq * fix: default subscribe to action_status * refactor: separate out async part * test: merge streams, but not data * test: merge streams with data * test: select from a stream * test: select from two streams * test: null after timeout * test: run them together * test: previous value after flush * doc: describe topic structure * wait 2s for output * test: use different port * test: await data with no push and change to push with qos 0 * test: similar inputs * test: fix port * doc: describe tests * test: renaming fields * test: publish back on bus * doc: testing publish back on bus * test: move to tests/bus.rs * refactor: feature gate bus * refactor: use `LinkRx::next()` * chore: use rumqtt main * ci: test all features * refactor: split joins out * refactor: redo without trait * fix: use the correct topics * feat: expose rumqttd console * feat: no route for unsubscribed actions * test: fail unregistered action * refactor: test/bus.rs * fix: `/subscriptions` might return empty list * fix: directly push `on_new_data` sequence numbers * test: fix port occupied * fix: `NoRoute` if no subscription only * test: addr occupied * style: note the unit in field name * test: fix joins * Revert "ci: test all features" This reverts commit 650c871. * Revert "feat: simulator as a service on bus" This reverts commit f8885ab. * style: name makes sense * style: name types * fix: include bus config
- Loading branch information
Devdutt Shenoi
authored
Sep 16, 2024
1 parent
0e9e6a9
commit ca07900
Showing
14 changed files
with
2,121 additions
and
33 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,201 @@ | ||
| use std::collections::HashMap; | ||
|
|
||
| use flume::{bounded, Receiver, Sender}; | ||
| use log::{error, warn}; | ||
| use serde_json::{json, Map, Value}; | ||
| use tokio::{select, task::JoinSet, time::interval}; | ||
|
|
||
| use crate::{ | ||
| base::{ | ||
| bridge::{BridgeTx, Payload}, | ||
| clock, | ||
| }, | ||
| config::{Field, JoinConfig, NoDataAction, PushInterval, SelectConfig}, | ||
| }; | ||
|
|
||
| type Json = Map<String, Value>; | ||
| type InputStream = String; | ||
| type FieldName = String; | ||
|
|
||
| pub struct Router { | ||
| map: HashMap<InputStream, Vec<Sender<(InputStream, Json)>>>, | ||
| pub tasks: JoinSet<()>, | ||
| } | ||
|
|
||
| impl Router { | ||
| pub async fn new( | ||
| configs: Vec<JoinConfig>, | ||
| bridge_tx: BridgeTx, | ||
| back_tx: Sender<Payload>, | ||
| ) -> Self { | ||
| let mut map: HashMap<InputStream, Vec<Sender<(InputStream, Json)>>> = HashMap::new(); | ||
| let mut tasks = JoinSet::new(); | ||
| for config in configs { | ||
| let (tx, rx) = bounded(1); | ||
| let mut fields = HashMap::new(); | ||
| for stream in &config.construct_from { | ||
| if let SelectConfig::Fields(selected_fields) = &stream.select_fields { | ||
| let renames: &mut HashMap<FieldName, Field> = | ||
| fields.entry(stream.input_stream.to_owned()).or_default(); | ||
| for field in selected_fields { | ||
| renames.insert(field.original.to_owned(), field.to_owned()); | ||
| } | ||
| } | ||
| if let Some(senders) = map.get_mut(&stream.input_stream) { | ||
| senders.push(tx.clone()); | ||
| continue; | ||
| } | ||
| map.insert(stream.input_stream.to_owned(), vec![tx.clone()]); | ||
| } | ||
| let joiner = Joiner { | ||
| rx, | ||
| joined: Json::new(), | ||
| config, | ||
| tx: bridge_tx.clone(), | ||
| fields, | ||
| back_tx: back_tx.clone(), | ||
| sequence: 0, | ||
| }; | ||
| tasks.spawn(joiner.start()); | ||
| } | ||
|
|
||
| Router { map, tasks } | ||
| } | ||
|
|
||
| pub async fn map(&mut self, input_stream: InputStream, json: Json) { | ||
| let Some(iter) = self.map.get(&input_stream) else { return }; | ||
| for tx in iter { | ||
| _ = tx.send_async((input_stream.clone(), json.clone())).await; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| struct Joiner { | ||
| rx: Receiver<(InputStream, Json)>, | ||
| joined: Json, | ||
| config: JoinConfig, | ||
| fields: HashMap<InputStream, HashMap<FieldName, Field>>, | ||
| tx: BridgeTx, | ||
| back_tx: Sender<Payload>, | ||
| sequence: u32, | ||
| } | ||
|
|
||
| impl Joiner { | ||
| async fn start(mut self) { | ||
| let PushInterval::OnTimeout(period) = self.config.push_interval_s else { | ||
| loop { | ||
| match self.rx.recv_async().await { | ||
| Ok((input_stream, json)) => self.update(input_stream, json), | ||
| Err(e) => { | ||
| error!("{e}"); | ||
| return; | ||
| } | ||
| } | ||
| self.send_data().await; | ||
| } | ||
| }; | ||
| let mut ticker = interval(period); | ||
| loop { | ||
| select! { | ||
| r = self.rx.recv_async() => { | ||
| match r { | ||
| Ok((input_stream, json)) => self.update(input_stream, json), | ||
| Err(e) => { | ||
| error!("{e}"); | ||
| return; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| _ = ticker.tick() => { | ||
| self.send_data().await | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Use data sequence and timestamp if data is to be pushed instantly | ||
| fn is_insertable(&self, key: &str) -> bool { | ||
| match key { | ||
| "timestamp" | "sequence" => self.config.push_interval_s == PushInterval::OnNewData, | ||
| _ => true, | ||
| } | ||
| } | ||
|
|
||
| fn update(&mut self, input_stream: InputStream, json: Json) { | ||
| if let Some(map) = self.fields.get(&input_stream) { | ||
| for (mut key, value) in json { | ||
| // drop unenumerated keys from json | ||
| let Some(field) = map.get(&key) else { continue }; | ||
| if let Some(name) = &field.renamed { | ||
| name.clone_into(&mut key); | ||
| } | ||
|
|
||
| if self.is_insertable(&key) { | ||
| self.joined.insert(key, value); | ||
| } | ||
| } | ||
| } else { | ||
| // Select All if no mapping exists | ||
| for (key, value) in json { | ||
| if self.is_insertable(&key) { | ||
| self.joined.insert(key, value); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| async fn send_data(&mut self) { | ||
| if self.joined.is_empty() { | ||
| return; | ||
| } | ||
|
|
||
| // timestamp and sequence values should be passed as is for instant push, else use generated values | ||
| let timestamp = self | ||
| .joined | ||
| .remove("timestamp") | ||
| .and_then(|value| { | ||
| value.as_i64().map_or_else( | ||
| || { | ||
| warn!( | ||
| "timestamp: {value:?} has unexpected type; defaulting to system time" | ||
| ); | ||
| None | ||
| }, | ||
| |v| Some(v as u64), | ||
| ) | ||
| }) | ||
| .unwrap_or_else(|| clock() as u64); | ||
| let sequence = self | ||
| .joined | ||
| .remove("sequence") | ||
| .and_then(|value| { | ||
| value.as_i64().map_or_else( | ||
| || { | ||
| warn!( | ||
| "sequence: {value:?} has unexpected type; defaulting to internal sequence" | ||
| ); | ||
| None | ||
| }, | ||
| |v| Some(v as u32), | ||
| ) | ||
| }) | ||
| .unwrap_or_else(|| { | ||
| self.sequence += 1; | ||
| self.sequence | ||
| }); | ||
| let payload = Payload { | ||
| stream: self.config.name.clone(), | ||
| sequence, | ||
| timestamp, | ||
| payload: json!(self.joined), | ||
| }; | ||
| if self.config.publish_on_service_bus { | ||
| _ = self.back_tx.send_async(payload.clone()).await; | ||
| } | ||
| self.tx.send_payload(payload).await; | ||
| if self.config.no_data_action == NoDataAction::Null { | ||
| self.joined.clear(); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.