Merge remote-tracking branch 'origin/main' into bus

.github/workflows/release.yml

@@ -23,13 +23,12 @@ jobs:
           target: x86_64-unknown-linux-musl
 
       - name: Build binary
-        working-directory: tools/${{ matrix.build }}
-        run: cargo build --verbose --release --target x86_64-unknown-linux-musl
+        run: cargo build -p ${{ matrix.build }} --verbose --release --target x86_64-unknown-linux-musl
 
       - name: Upload release archive
         uses: softprops/action-gh-release@v1
         with:
-          files: tools/${{ matrix.build }}/target/x86_64-unknown-linux-musl/release/${{ matrix.build }}
+          files: target/x86_64-unknown-linux-musl/release/${{ matrix.build }}
 
   build-release:
     name: Build release for ${{ matrix.target }}

Cargo.toml

@@ -3,17 +3,8 @@ serial = { git = "https://github.com/bytebeamio/serial-rs", branch = "android_fi
 
 [workspace]
 resolver = "2"
-members = [
-    "uplink",
-    "storage",
-    "tools/utils",
-]
-exclude = [
-    "tools/deserialize-backup",
-    "tools/simulator",
-    "tools/system-stats",
-    "tools/tunshell",
-]
+members = ["uplink", "storage", "tools/utils", "tools/system-stats"]
+exclude = ["tools/deserialize-backup", "tools/simulator", "tools/tunshell"]
 
 [workspace.dependencies]
 bytes = "1"

configs/config.toml

@@ -17,6 +17,9 @@ persistence_path = "/tmp/uplink/"
 # Size of in-memory buffer for dynamically created streams. Used for backlog management.
 default_buf_size = 1024 # 1KB
 
+# Maximum number of data streams that can be accepted by uplink
+max_stream_count = 10
+
 # MQTT client configuration
 #
 # Required Parameters
@@ -168,16 +171,14 @@ update_period = 30
 enabled = true
 port = 3333
 
-# Configurations associated with running uplink in simulator mode, if enabled
-# uplink will push simulated data for device_id 1..=num_devices and respond to any
-# actions triggered on the tenant connected to, with progress 0..=100.
-# NOTE: uplink bridge will hence be turned off.
+# Configurations associated with running uplink in simulator mode, if enabled uplink
+# will push simulated data for respond to actions listed with progress 0..=100.
 #
 # Required Parameters
-# - num_devices: number of devices to be simulated
+# - actions: action routes on which uplink simulator can accept actions
 # - gps_paths: path to directory containing files with gps paths to be used in simulation
 # [simulator]
-# num_devices = 10
+# actions = [{ name = "lock" }]
 # gps_paths = "./paths/"
 
 # Configuration of logger, journalctl in the case of linux. As configured, uplink will

rust-toolchain.toml

@@ -1,12 +1,10 @@
 [toolchain]
-channel = "1.78.0"
+channel = "1.80.0"
 targets = [
     "x86_64-linux-android",
     "i686-linux-android",
     "aarch64-linux-android",
-    "armv7-linux-androideabi"
+    "armv7-linux-androideabi",
 ]
 profile = "minimal"
-components = [
-    "clippy"
-]
+components = ["clippy"]

tools/system-stats/Cargo.toml

@@ -5,14 +5,14 @@ edition = "2021"
 authors = ["Devdutt Shenoi <[email protected]>"]
 
 [dependencies]
-futures-util = { version = "0.3", features = ["sink"] }
-log = "0.4"
-serde = { version = "1", features = ["derive"] }
-serde_json = "1.0"
+futures-util = { workspace = true, features = ["sink"] }
+log = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
 simplelog = "0.12.0"
 structopt = "0.3"
 sysinfo = "0.26"
-thiserror = "1"
-tokio = { version = "1", features = ["rt-multi-thread", "macros", "net"] }
-tokio-stream = "0.1"
-tokio-util = { version = "0.7", features = ["codec", "time"] }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["macros", "net"] }
+tokio-stream = { workspace = true }
+tokio-util = { workspace = true }

tools/system-stats/src/lib.rs

@@ -59,10 +59,7 @@ pub struct System {
 impl System {
     fn init(sys: &sysinfo::System) -> System {
         System {
-            kernel_version: match sys.kernel_version() {
-                Some(kv) => kv,
-                None => String::default(),
-            },
+            kernel_version: sys.kernel_version().unwrap_or_default(),
             total_memory: sys.total_memory(),
             ..Default::default()
         }
@@ -622,7 +619,7 @@ impl StatCollector {
         self.sys.refresh_processes();
         let timestamp = clock();
         for (&id, p) in self.sys.processes() {
-            let name = p.cmd().get(0).map(|s| s.to_string()).unwrap_or(p.name().to_string());
+            let name = p.cmd().first().map(|s| s.to_string()).unwrap_or(p.name().to_string());
 
             if self.config.process_names.contains(&name) {
                 let payload = self.processes.push(id.as_u32(), p, name, timestamp);

uplink/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "uplink"
-version = "2.14.0"
+version = "2.14.1"
 authors = ["tekjar <[email protected]>"]
 edition = "2021"
 

uplink/src/base/bridge/actions_lane.rs

@@ -96,7 +96,7 @@ impl ActionsBridge {
         action_status.batch_size = 1;
 
         streams_config.insert("action_status".to_owned(), action_status);
-        let mut streams = Streams::new(config.clone(), device_config, package_tx, metrics_tx);
+        let mut streams = Streams::new(1, device_config, package_tx, metrics_tx);
         streams.config_streams(streams_config);
 
         Self {

uplink/src/base/bridge/data_lane.rs

@@ -37,7 +37,8 @@ impl DataBridge {
         let (data_tx, data_rx) = bounded(10);
         let (ctrl_tx, ctrl_rx) = bounded(1);
 
-        let mut streams = Streams::new(config.clone(), device_config, package_tx, metrics_tx);
+        let mut streams =
+            Streams::new(config.max_stream_count, device_config, package_tx, metrics_tx);
         streams.config_streams(config.streams.clone());
 
         Self { data_tx, data_rx, config, streams, ctrl_rx, ctrl_tx }

uplink/src/base/bridge/streams.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use flume::Sender;
 use log::{error, info, trace};
 
-use crate::config::{Config, DeviceConfig, StreamConfig};
+use crate::config::{DeviceConfig, StreamConfig};
 
 use super::{
     delaymap::DelayMap,
@@ -13,7 +13,7 @@ use super::{
 };
 
 pub struct Streams<T> {
-    config: Arc<Config>,
+    max_stream_count: usize,
     device_config: Arc<DeviceConfig>,
     data_tx: Sender<Box<dyn Package>>,
     metrics_tx: Sender<StreamMetrics>,
@@ -23,17 +23,18 @@ pub struct Streams<T> {
 
 impl<T: Point> Streams<T> {
     pub fn new(
-        config: Arc<Config>,
+        max_stream_count: usize,
         device_config: Arc<DeviceConfig>,
         data_tx: Sender<Box<dyn Package>>,
         metrics_tx: Sender<StreamMetrics>,
     ) -> Self {
+        let map = HashMap::with_capacity(max_stream_count);
         Self {
-            config,
+            max_stream_count,
             device_config,
             data_tx,
             metrics_tx,
-            map: HashMap::new(),
+            map,
             stream_timeouts: DelayMap::new(),
         }
     }
@@ -50,8 +51,11 @@ impl<T: Point> Streams<T> {
 
         // Create stream if it doesn't already exist
         if !self.map.contains_key(&stream_name) {
-            if self.config.simulator.is_none() && self.map.keys().len() > 20 {
-                error!("Failed to create {:?} stream. More than max 20 streams", stream_name);
+            if self.map.keys().len() > self.max_stream_count {
+                error!(
+                    "Failed to create {:?} stream. More than max {} streams",
+                    stream_name, self.max_stream_count
+                );
                 return;
             }