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

.github/workflows/release.yml

@@ -96,11 +96,39 @@ jobs:
       - name: Rename ${{ matrix.target }} binary
         if: matrix.build != 'windows'
         run: mv "target/${{ matrix.target }}/release/uplink" "uplink-${{ matrix.target }}"
-      
+
       - name: Rename ${{ matrix.target }} binary
         if: matrix.build == 'windows'
         run: mv "target/${{ matrix.target }}/release/uplink.exe" "uplink-${{ matrix.target }}.exe"
-
+
+      - name: Upload release archive
+        uses: softprops/action-gh-release@v1
+        with:
+          files: uplink*
+  build-release-android:
+    name: Build release for android
+    runs-on: ubuntu-latest
+    container:
+      image: bytebeamio/rust-android
+    strategy:
+      fail-fast: false
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: perms
+        run: chown root:root .
+
+      - name: Build for armv7
+        run: cargo ndk --target armv7-linux-androideabi --platform 23 build --release --bin uplink
+
+      - name: Build for aarch64
+        run: cargo ndk --target aarch64-linux-android --platform 23 build --release --bin uplink
+
+      - name: Rename uplink binaries
+        run: |
+          mv "target/aarch64-linux-android/release/uplink" "uplink-aarch64-linux-android"
+          mv "target/armv7-linux-androideabi/release/uplink" "uplink-armv7-linux-androideabi"
+
       - name: Upload release archive
         uses: softprops/action-gh-release@v1
         with:

Cargo.toml

@@ -5,13 +5,13 @@ serial = { git = "https://github.com/bytebeamio/serial-rs", branch = "android_fi
 resolver = "2"
 members = [
     "uplink",
-    "storage"
+    "storage",
+    "tools/utils",
 ]
 exclude = [
     "tools/deserialize-backup",
     "tools/simulator",
     "tools/tunshell",
-    "tools/utils",
 ]
 
 [profile.dev]

README.md

@@ -14,7 +14,6 @@ Uplink is a rust based utility for efficiently sending data and receiving comman
 - Provides remote shell access through [Tunshell][tunshell]
 - Supports TLS with easy cross-compilation.
 
-
 ### Quickstart
 
 #### Install Uplink

examples/android/README.md

@@ -0,0 +1,3 @@
+## Uplink example setup for android
+
+The `payload` directory has an example uplink setup that can be installed in an android device.

examples/android/VERSION.txt

@@ -0,0 +1 @@
+r2-2023-10-13-NOWD-AOSP

examples/android/env.sh

@@ -0,0 +1,2 @@
+export DATA_DIR=/data/local/tmp/uplink
+export UPLINK_LOG_LEVEL=-v

examples/android/etc/uplink.config.toml

@@ -0,0 +1,39 @@
+processes = []
+action_redirections = { "update_firmware" = "install_update", "send_script" = "run_script" }
+script_runner = [{ name = "run_script" }]
+persistence_path = "/data/local/tmp/uplink/persistence"
+
+[tcpapps.app]
+port = 8031
+#actions = [{ name = "install_update" }]
+
+[tcpapps.app2]
+port = 8032
+
+[system_stats]
+enabled = true
+process_names = ["/data/local/uplinkmodule/bin/uplink", "com.foobnix.pro.pdf.reader"]
+update_period = 2
+stream_size = 1
+
+[ota_installer]
+path="/data/local/tmp/uplink/installer"
+actions=[{name="install_update", timeout=310}]
+uplink_port=8032
+
+[downloader]
+actions = [{ name = "update_firmware" }, { name = "send_script" }]
+path = "/data/local/tmp/uplink/downloader"
+
+[streams.device_shadow]
+topic = "/tenants/{tenant_id}/devices/{device_id}/events/device_shadow/jsonarray"
+buf_size = 128
+flush_period = 30
+
+[logging]
+tags = ["*"]
+min_level = 4
+
+[console]
+enabled = true
+port = 9328

examples/android/service.sh

@@ -0,0 +1,11 @@
+#!/system/bin/sh
+
+set -x
+
+export MODULE_DIR="$( cd -- "$(dirname "$0")" >/dev/null 2>&1 ; pwd -P )"
+. $MODULE_DIR/env.sh
+mkdir -p $DATA_DIR
+
+cd $DATA_DIR || exit
+
+$MODULE_DIR/bin/uplink -a $DATA_DIR/device.json -c $MODULE_DIR/etc/uplink.config.toml $UPLINK_LOG_LEVEL 2>&1 | $MODULE_DIR/bin/logrotate --max-size 10000000 --backup-files-count 30 --output $DATA_DIR/out.log

examples/demo.cpp

@@ -0,0 +1,130 @@
+#include <iostream>
+#include <sys/socket.h>
+#include <unistd.h>
+#include <string>
+#include <thread>
+#include <sstream>
+#include <vector>
+#include <sys/time.h>
+#include <netdb.h>
+#include <string.h>
+#include <mutex>
+#include <fcntl.h>
+#include <optional>
+
+using namespace std;
+
+optional<string> readUntilNewline(int socketFD) {
+    std::vector<char> buffer;
+    char c;
+    while (true) {
+        int res = read(socketFD, &c, 1);
+        if (res == 0) {
+            // socket is in blocking mode, read return value 0 means socket has closed
+            return {};
+        }
+        if (c == '\n') {
+            break;
+        }
+        buffer.push_back(c);
+    }
+    return string(buffer.begin(), buffer.end());
+}
+
+ssize_t writeToSocket(int socketFD, const string &data) {
+    return send(socketFD, data.c_str(), data.length(), MSG_NOSIGNAL);
+}
+
+long current_time_millis() {
+  struct timeval tv;
+  gettimeofday(&tv, NULL);
+  long long millis =
+      ((long long)tv.tv_sec) * 1000 + ((long long)tv.tv_usec) / 1000;
+  return millis;
+}
+
+int connectToUplink(int port) {
+    struct addrinfo hints, *res;
+    memset(&hints, 0, sizeof hints);
+    hints.ai_family = AF_UNSPEC;
+    hints.ai_socktype = SOCK_STREAM;
+
+    if (getaddrinfo("127.0.0.1", std::to_string(port).c_str(), &hints, &res) != 0) {
+        cout << "getaddrinfo failed for port " << port << endl;
+        return 0;
+    }
+
+    int socketFD = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+
+    fcntl(socketFD, F_SETFL, O_NONBLOCK);
+
+    fd_set fdset;
+    struct timeval tv;
+    FD_ZERO(&fdset);
+    FD_SET(socketFD, &fdset);
+    tv.tv_sec = 0;
+    tv.tv_usec = 250000; // 250 ms connection timeout
+
+    ::connect(socketFD, res->ai_addr, res->ai_addrlen);
+
+    if (select(socketFD + 1, nullptr, &fdset, nullptr, &tv) == 1) {
+        int so_error;
+        socklen_t len = sizeof so_error;
+
+        getsockopt(socketFD, SOL_SOCKET, SO_ERROR, &so_error, &len);
+
+        if (so_error == 0) {
+            freeaddrinfo(res);
+            int flags = fcntl(socketFD, F_GETFL, 0);
+            flags = flags & ~O_NONBLOCK;
+            fcntl(socketFD, F_SETFL, flags);
+            return socketFD;
+        } else {
+            cout << "getsockopt returned error " << so_error << endl;
+            return 0;
+        }
+    }
+
+    cout << "select failed for port " << port << endl;
+    freeaddrinfo(res);
+    return 0;
+}
+
+int main() {
+    int socketFD = connectToUplink(8031);
+    std::mutex clientLock;
+    if (socketFD == 0) {
+        cout << "couldn't connect" << endl;
+        return 1;
+    }
+
+    std::thread actionResponder([&] () {
+        while (true) {
+            optional<string> lineO = readUntilNewline(socketFD);
+            if (!lineO.has_value()) {
+                break;
+            }
+            string line = lineO.value();
+            clientLock.lock();
+            cout << line << endl;
+            // push action_status for this action
+            clientLock.unlock();
+        }
+        cout << "stopping responder thread" << endl;
+    });
+
+    int sequence = 1;
+    while (true) {
+        sleep(1);
+        ostringstream oss;
+        oss << "{\"stream\": \"device_shadow\", \"sequence\": " << sequence++ << ", \"timestamp\": " << current_time_millis() << ", \"a\": 1, \"b\": 2 }" << endl;
+        clientLock.lock();
+        if (writeToSocket(socketFD, oss.str()) == -1) {
+            break;
+        }
+        clientLock.unlock();
+    }
+    actionResponder.join();
+    cout << "stopping pusher thread" << endl;
+    return 0;
+}

rust-toolchain.toml

@@ -0,0 +1,12 @@
+[toolchain]
+channel = "nightly-2023-06-15"
+targets = [
+    "x86_64-linux-android",
+    "i686-linux-android",
+    "aarch64-linux-android",
+    "armv7-linux-androideabi"
+]
+profile = "minimal"
+components = [
+    "clippy"
+]

uplink/Cargo.toml

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

uplink/src/base/actions.rs

@@ -1,5 +1,4 @@
 use serde::{Deserialize, Serialize};
-use serde_json::json;
 use tokio::time::Instant;
 
 use crate::{Payload, Point};
@@ -98,33 +97,17 @@ impl ActionResponse {
         self
     }
 
-    pub fn as_payload(&self) -> Payload {
-        Payload::from(self)
-    }
-
     pub fn from_payload(payload: &Payload) -> Result<Self, serde_json::Error> {
         let intermediate = serde_json::to_value(payload)?;
         serde_json::from_value(intermediate)
     }
 }
 
-impl From<&ActionResponse> for Payload {
-    fn from(resp: &ActionResponse) -> Self {
-        Self {
-            stream: "action_status".to_owned(),
-            sequence: resp.sequence,
-            timestamp: resp.timestamp,
-            payload: json!({
-                "id": resp.action_id,
-                "state": resp.state,
-                "progress": resp.progress,
-                "errors": resp.errors
-            }),
-        }
+impl Point for ActionResponse {
+    fn stream_name(&self) -> &str {
+        "action_status"
     }
-}
 
-impl Point for ActionResponse {
     fn sequence(&self) -> u32 {
         self.sequence
     }