basic proof of concept

scylla-server-rust/.gitignore

@@ -22,3 +22,4 @@ prisma.rs
 
 # protobuf
 serverdata.rs
+command_data.rs

scylla-server-rust/Cargo.toml

@@ -22,6 +22,7 @@ rand = "0.8.5"
 console-subscriber = { version = "0.3.0", optional = true }
 ringbuffer = "0.15.0"
 clap = { version = "4.5.11", features = ["derive", "env"] }
+axum-extra = { version = "0.9.3", features = ["query"] }
 
 [features]
 top = ["dep:console-subscriber"]

scylla-server-rust/build.rs

@@ -8,6 +8,7 @@ fn main() {
         .includes(["src/proto"])
         // Inputs must reside in some of include paths.
         .input("src/proto/serverdata.proto")
+        .input("src/proto/command_data.proto")
         // Specify output directory relative to Cargo output directory.
         .out_dir("src")
         .run_from_script();

scylla-server-rust/src/controllers/mod.rs

@@ -5,3 +5,4 @@ pub mod location_controller;
 pub mod node_controller;
 pub mod run_controller;
 pub mod system_controller;
+pub mod send_config_controller;

scylla-server-rust/src/controllers/send_config_controller.rs

@@ -0,0 +1,49 @@
+use std::sync::Arc;
+
+use axum::{extract::Path, Extension};
+use axum_extra::extract::Query;
+use protobuf::Message;
+use rumqttc::v5::AsyncClient;
+use serde::Deserialize;
+use tracing::{info, warn};
+
+use crate::{command_data::CommandData, error::ScyllaError};
+
+#[derive(Deserialize, Debug)]
+pub struct ConfigRequest {
+    pub data: Vec<f32>,
+}
+
+pub async fn send_config(
+    Path(key): Path<String>,
+    data_query: Query<ConfigRequest>,
+    Extension(client): Extension<Option<Arc<AsyncClient>>>,
+) -> Result<(), ScyllaError> {
+    info!(
+        "Sending car config with key: {}, and values: {:?}",
+        key, data_query.0
+    );
+    let Some(client) = client else {
+        return Err(ScyllaError::NotProd);
+    };
+
+    let mut payload = CommandData::new();
+    payload.data = data_query.0.data;
+    let Ok(bytes) = payload.write_to_bytes() else {
+        return Err(ScyllaError::ImpossibleEncoding);
+    };
+
+    if let Err(err) = client
+        .publish(
+            format!("Calypso/Bidir/Command/{}", key),
+            rumqttc::v5::mqttbytes::QoS::ExactlyOnce,
+            false,
+            bytes,
+        )
+        .await
+    {
+        warn!("Could not publish instruction: {}", err);
+        return Err(ScyllaError::CommFailure);
+    }
+    Ok(())
+}

scylla-server-rust/src/error.rs

@@ -10,7 +10,14 @@ use tracing::warn;
 
 pub enum ScyllaError {
     PrismaError(QueryError),
+    /// A generic not found for a prisma query
     NotFound,
+    /// Not available in mock mode, which the system is in
+    NotProd,
+    /// An instruction was not encodable
+    ImpossibleEncoding,
+    /// Could not communicate to car
+    CommFailure,
 }
 
 impl From<QueryError> for ScyllaError {
@@ -32,6 +39,9 @@ impl IntoResponse for ScyllaError {
             }
             ScyllaError::PrismaError(_) => StatusCode::BAD_REQUEST,
             ScyllaError::NotFound => StatusCode::NOT_FOUND,
+            ScyllaError::NotProd => StatusCode::SERVICE_UNAVAILABLE,
+            ScyllaError::ImpossibleEncoding => StatusCode::UNPROCESSABLE_ENTITY,
+            ScyllaError::CommFailure => StatusCode::BAD_GATEWAY,
         };
 
         status.into_response()

scylla-server-rust/src/lib.rs

@@ -9,6 +9,7 @@ pub mod transformers;
 pub mod prisma;
 
 pub mod serverdata;
+pub mod command_data;
 
 /// The type descriptor of the database passed to the middlelayer through axum state
 pub type Database = std::sync::Arc<prisma::PrismaClient>;

scylla-server-rust/src/main.rs

@@ -7,10 +7,13 @@ use axum::{
 };
 use clap::Parser;
 use prisma_client_rust::chrono;
+use rumqttc::v5::AsyncClient;
 use scylla_server_rust::{
     controllers::{
         self, data_type_controller, driver_controller, location_controller, node_controller,
-        run_controller, system_controller,
+        run_controller,
+        send_config_controller::{self},
+        system_controller,
     },
     prisma::PrismaClient,
     processors::{
@@ -55,7 +58,12 @@ struct ScyllaArgs {
     siren_host_url: String,
 
     /// The time, in seconds between collection for a batch upsert
-    #[arg(short = 't', long, env = "SCYLLA_BATCH_UPSERT_TIME", default_value = "10")]
+    #[arg(
+        short = 't',
+        long,
+        env = "SCYLLA_BATCH_UPSERT_TIME",
+        default_value = "10"
+    )]
     batch_upsert_time: u64,
 }
 
@@ -138,11 +146,12 @@ async fn main() {
         token.clone(),
     ));
 
-    // if PROD_SCYLLA=false
-    if !cli.prod {
+    // if PROD_SCYLLA=false, also procur a client for use in the config state
+    let client: Option<Arc<AsyncClient>> = if !cli.prod {
         info!("Running processor in mock mode, no data will be stored");
         let recv = MockProcessor::new(io);
         tokio::spawn(recv.generate_mock());
+        None
     } else {
         // creates the initial run
         let curr_run = run_service::create_run(&db, chrono::offset::Utc::now().timestamp_millis())
@@ -153,15 +162,18 @@ async fn main() {
         // run prod if this isnt present
         // create and spawn the mqtt processor
         info!("Running processor in MQTT (production) mode");
-        let recv = MqttProcessor::new(
+        let (recv, opts) = MqttProcessor::new(
             mqtt_send,
             new_run_receive,
             cli.siren_host_url,
             curr_run.id,
             io,
         );
-        tokio::spawn(recv.process_mqtt());
-    }
+        let (client, eventloop) = AsyncClient::new(opts, 600);
+        let client_sharable: Arc<AsyncClient> = Arc::new(client);
+        tokio::spawn(recv.process_mqtt(client_sharable.clone(), eventloop));
+        Some(client_sharable)
+    };
 
     let app = Router::new()
         // DATA ROUTES
@@ -186,6 +198,8 @@ async fn main() {
         )
         // SYSTEMS
         .route("/systems", get(system_controller::get_all_systems))
+        // CONFIG
+        .route("/config/:key", post(send_config_controller::send_config).layer(Extension(client)))
         // for CORS handling
         .layer(
             CorsLayer::new()

scylla-server-rust/src/mod.rs

@@ -1,3 +1,4 @@
 // @generated
 
+pub mod command_data;
 pub mod serverdata;

scylla-server-rust/src/processors/mqtt_processor.rs

@@ -1,12 +1,12 @@
 use core::fmt;
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use prisma_client_rust::{bigdecimal::ToPrimitive, chrono, serde_json};
 use protobuf::Message;
 use ringbuffer::RingBuffer;
 use rumqttc::v5::{
     mqttbytes::v5::{LastWill, Packet, Publish},
-    AsyncClient, Event, MqttOptions,
+    AsyncClient, Event, EventLoop, MqttOptions,
 };
 use socketioxide::SocketIo;
 use tokio::sync::mpsc::{Receiver, Sender};
@@ -22,7 +22,6 @@ pub struct MqttProcessor {
     new_run_channel: Receiver<run_service::public_run::Data>,
     curr_run: i32,
     io: SocketIo,
-    mqtt_ops: MqttOptions,
 }
 
 impl MqttProcessor {
@@ -32,16 +31,16 @@ impl MqttProcessor {
     /// * `db` - The database to store the data in
     /// * `io` - The socketio layer to send the data to
     ///
-    /// Returns the instance and the event loop, which can be passed into the process_mqtt func to begin recieiving
+    /// Returns the instance and options to create a client, which is then used in the process_mqtt loop
     pub fn new(
         channel: Sender<ClientData>,
         new_run_channel: Receiver<run_service::public_run::Data>,
         mqtt_path: String,
         initial_run: i32,
         io: SocketIo,
-    ) -> MqttProcessor {
+    ) -> (MqttProcessor, MqttOptions) {
         // create the mqtt client and configure it
-        let mut create_opts = MqttOptions::new(
+        let mut mqtt_opts = MqttOptions::new(
             "ScyllaServer",
             mqtt_path.split_once(':').expect("Invalid Siren URL").0,
             mqtt_path
@@ -51,7 +50,7 @@ impl MqttProcessor {
                 .parse::<u16>()
                 .expect("Invalid Siren port"),
         );
-        create_opts
+        mqtt_opts
             .set_last_will(LastWill::new(
                 "Scylla/Status",
                 "Scylla has disconnected!",
@@ -65,27 +64,28 @@ impl MqttProcessor {
             .set_session_expiry_interval(Some(u32::MAX))
             .set_topic_alias_max(Some(600));
 
-        MqttProcessor {
-            channel,
-            new_run_channel,
-            curr_run: initial_run,
-            io,
-            mqtt_ops: create_opts,
-        }
+        // TODO mess with incoming message cap if db, etc. cannot keep up
+
+        (
+            MqttProcessor {
+                channel,
+                new_run_channel,
+                curr_run: initial_run,
+                io,
+            },
+            mqtt_opts,
+        )
     }
 
     /// This handles the reception of mqtt messages, will not return
-    /// * `connect` - The eventloop returned by ::new to connect to
-    pub async fn process_mqtt(mut self) {
+    /// * `eventloop` - The eventloop returned by ::new to connect to.  The loop isnt sync so this is the best that can be done
+    /// * `client` - The async mqttt v5 client to use for subscriptions
+    pub async fn process_mqtt(mut self, client: Arc<AsyncClient>, mut eventloop: EventLoop) {
         let mut view_interval = tokio::time::interval(Duration::from_secs(3));
 
         let mut latency_interval = tokio::time::interval(Duration::from_millis(250));
         let mut latency_ringbuffer = ringbuffer::AllocRingBuffer::<i64>::new(20);
 
-        // process over messages, non blocking
-        // TODO mess with incoming message cap if db, etc. cannot keep up
-        let (client, mut connect) = AsyncClient::new(self.mqtt_ops.clone(), 600);
-
         debug!("Subscribing to siren");
         client
             .subscribe("#", rumqttc::v5::mqttbytes::QoS::ExactlyOnce)
@@ -95,7 +95,7 @@ impl MqttProcessor {
         loop {
             #[rustfmt::skip] // rust cannot format this macro for some reason
             tokio::select! {
-                msg = connect.poll() => match msg {
+                msg = eventloop.poll() => match msg {
                     Ok(Event::Incoming(Packet::Publish(msg))) => {
                         trace!("Received mqtt message: {:?}", msg);
                         // parse the message into the data and the node name it falls under

scylla-server-rust/src/proto/command_data.proto

@@ -0,0 +1,7 @@
+syntax = "proto3";
+
+package command_data.v1;
+
+message CommandData {
+   repeated float data = 1;
+}