Disable old wrap REST code and pass tx and sensor_channel to new context

Author: manio

src/ev.rs

@@ -1,9 +1,7 @@
 use simplelog::*;
 use std::path::PathBuf;
-use std::sync::Arc;
 use tokio::fs;
 use tokio::sync::mpsc::Sender;
-use tokio::sync::Mutex;
 
 // protobuf stuff:
 include!(concat!(env!("OUT_DIR"), "/protos/mod.rs"));
@@ -14,7 +12,6 @@ use crate::mitm::{ENCRYPTED, FRAME_TYPE_FIRST, FRAME_TYPE_LAST};
 use protobuf::Message;
 
 use serde::Deserialize;
-use warp::Filter;
 
 pub static FORD_EV_MODEL: &[u8] = include_bytes!("protos/ford_ev_model.bin");
 pub const EV_MODEL_FILE: &str = "/etc/aa-proxy-rs/ev_model.bin";
@@ -31,68 +28,6 @@ pub struct BatteryData {
     pub battery_level: f32,
 }
 
-// reset server context
-#[derive(Clone)]
-pub struct RestContext {
-    pub sensor_channel: Option<u8>,
-    pub ev_battery_capacity: u64,
-    pub ev_factor: f32,
-}
-
-pub async fn rest_server(tx: Sender<Packet>, ctx: Arc<Mutex<RestContext>>) -> Result<()> {
-    let battery_route = warp::post()
-        .and(warp::path("battery"))
-        .and(warp::body::json())
-        .and(warp::any().map({
-            let ctx = ctx.clone();
-            move || ctx.clone()
-        }))
-        .and_then(move |data: BatteryData, ctx: Arc<Mutex<RestContext>>| {
-            let tx = tx.clone();
-            async move {
-                if data.battery_level < 0.0 || data.battery_level > 100.0 {
-                    let msg = format!(
-                        "battery_level out of range: {} (expected 0.0–100.0)",
-                        data.battery_level
-                    );
-                    return Ok::<_, warp::Rejection>(warp::reply::with_status(
-                        msg,
-                        warp::http::StatusCode::BAD_REQUEST,
-                    ));
-                }
-
-                info!("{} Received battery level: {}", NAME, data.battery_level);
-                let rest_ctx = ctx.lock().await;
-                if let Some(ch) = rest_ctx.sensor_channel {
-                    if let Err(e) = send_ev_data(
-                        tx,
-                        data.battery_level,
-                        ch,
-                        rest_ctx.ev_battery_capacity,
-                        rest_ctx.ev_factor,
-                    )
-                    .await
-                    {
-                        error!("{} EV model error: {}", NAME, e);
-                    }
-                } else {
-                    warn!("{} Not sending packet because no sensor channel yet", NAME);
-                }
-
-                Ok(warp::reply::with_status(
-                    "OK".into(),
-                    warp::http::StatusCode::OK,
-                ))
-            }
-        });
-
-    info!("{} Server running on http://127.0.0.1:3030", NAME);
-
-    warp::serve(battery_route).run(([127, 0, 0, 1], 3030)).await;
-
-    Ok(())
-}
-
 fn scale_percent_to_value(percent: f32, max_value: u64) -> u64 {
     let scaled = (percent as f64 / 100.0) * max_value as f64;
     scaled.round() as u64

src/io_uring.rs

@@ -33,7 +33,6 @@ pub const BUFFER_LEN: usize = 16 * 1024;
 const TCP_CLIENT_TIMEOUT: Duration = Duration::new(30, 0);
 
 use crate::config::SharedConfig;
-use crate::ev::{rest_server, RestContext};
 use crate::mitm::endpoint_reader;
 use crate::mitm::proxy;
 use crate::mitm::Packet;
@@ -210,6 +209,8 @@ pub async fn io_loop(
     need_restart: Arc<Notify>,
     tcp_start: Arc<Notify>,
     config: SharedConfig,
+    tx: Arc<Mutex<Option<Sender<Packet>>>>,
+    sensor_channel: Arc<Mutex<Option<u8>>>,
 ) -> Result<()> {
     // prepare/bind needed TCP listeners
     let mut dhu_listener = None;
@@ -367,19 +368,9 @@ pub async fn io_loop(
         }
 
         // handling battery in JSON
-        let mut rest_server_handle = None;
-        let mut rest_ctx = None;
         if config.mitm && config.ev {
-            let ctx = RestContext {
-                sensor_channel: None,
-                ev_battery_capacity: config.ev_battery_capacity,
-                ev_factor: config.ev_factor,
-            };
-            let ctx = Arc::new(Mutex::new(ctx));
-
-            let tx = tx_hu.clone();
-            rest_server_handle = Some(tokio::spawn(rest_server(tx, ctx.clone())));
-            rest_ctx = Some(ctx);
+            let mut tx_lock = tx.lock().await;
+            *tx_lock = Some(tx_hu.clone());
         }
 
         // dedicated reading threads:
@@ -394,7 +385,7 @@ pub async fn io_loop(
             rx_hu,
             rxr_md,
             config.clone(),
-            rest_ctx.clone(),
+            sensor_channel.clone(),
         ));
         from_stream = tokio_uring::spawn(proxy(
             ProxyType::MobileDevice,
@@ -404,7 +395,7 @@ pub async fn io_loop(
             rx_md,
             rxr_hu,
             config.clone(),
-            rest_ctx.clone(),
+            sensor_channel.clone(),
         ));
 
         // Thread for monitoring transfer
@@ -439,9 +430,12 @@ pub async fn io_loop(
         from_file.abort();
         from_stream.abort();
         monitor.abort();
-        if let Some(handle) = rest_server_handle {
-            handle.abort();
-        }
+
+        // set webserver context EV stuff to None
+        let mut tx_lock = tx.lock().await;
+        *tx_lock = None;
+        let mut sc_lock = sensor_channel.lock().await;
+        *sc_lock = None;
         // stop EV battery logger if neded
         if let Some(ref path) = config.ev_battery_logger {
             let _ = Command::new(path).arg("stop").spawn();

src/main.rs

@@ -10,6 +10,7 @@ mod web;
 
 use crate::config::AppConfig;
 use crate::config::SharedConfig;
+use crate::mitm::Packet;
 use bluetooth::bluetooth_setup_connection;
 use bluetooth::bluetooth_stop;
 use clap::Parser;
@@ -25,6 +26,8 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::runtime::Builder;
+use tokio::sync::mpsc::Sender;
+use tokio::sync::Mutex;
 use tokio::sync::Notify;
 use tokio::time::Instant;
 
@@ -162,6 +165,8 @@ async fn tokio_main(
     need_restart: Arc<Notify>,
     tcp_start: Arc<Notify>,
     config_file: PathBuf,
+    tx: Arc<Mutex<Option<Sender<Packet>>>>,
+    sensor_channel: Arc<Mutex<Option<u8>>>,
 ) {
     let accessory_started = Arc::new(Notify::new());
     let accessory_started_cloned = accessory_started.clone();
@@ -171,8 +176,8 @@ async fn tokio_main(
         let state = web::AppState {
             config: config.clone(),
             config_file: config_file.into(),
-            tx: None,
-            sensor_channel: None.into(),
+            tx,
+            sensor_channel,
         };
         let app = web::app(state.into());
 
@@ -326,15 +331,33 @@ fn main() {
     let tcp_start_cloned = tcp_start.clone();
     let config = Arc::new(RwLock::new(config));
     let config_cloned = config.clone();
+    let tx = Arc::new(Mutex::new(None));
+    let tx_cloned = tx.clone();
+    let sensor_channel = Arc::new(Mutex::new(None));
+    let sensor_channel_cloned = sensor_channel.clone();
 
     // build and spawn main tokio runtime
     let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
     runtime.spawn(async move {
-        tokio_main(config_cloned, need_restart, tcp_start, args.config.clone()).await
+        tokio_main(
+            config_cloned,
+            need_restart,
+            tcp_start,
+            args.config.clone(),
+            tx_cloned,
+            sensor_channel_cloned,
+        )
+        .await
     });
 
     // start tokio_uring runtime simultaneously
-    let _ = tokio_uring::start(io_loop(need_restart_cloned, tcp_start_cloned, config));
+    let _ = tokio_uring::start(io_loop(
+        need_restart_cloned,
+        tcp_start_cloned,
+        config,
+        tx,
+        sensor_channel,
+    ));
 
     info!(
         "🚩 aa-proxy-rs terminated, running time: {}",

src/mitm.rs

@@ -29,7 +29,6 @@ use protos::ControlMessageType::{self, *};
 
 use crate::config::AppConfig;
 use crate::config::HexdumpLevel;
-use crate::ev::RestContext;
 use crate::io_uring::Endpoint;
 use crate::io_uring::IoDevice;
 use crate::io_uring::BUFFER_LEN;
@@ -271,7 +270,7 @@ pub async fn pkt_modify_hook(
     video_in_motion: bool,
     ev: bool,
     ctx: &mut ModifyContext,
-    rest_ctx: Option<Arc<tokio::sync::Mutex<RestContext>>>,
+    sensor_channel: Arc<tokio::sync::Mutex<Option<u8>>>,
     ev_battery_logger: Option<PathBuf>,
 ) -> Result<bool> {
     // if for some reason we have too small packet, bail out
@@ -413,10 +412,8 @@ pub async fn pkt_modify_hook(
                     // set in local context
                     ctx.sensor_channel = Some(svc.id() as u8);
                     // set in REST server context for remote EV requests
-                    if let Some(ctx) = rest_ctx {
-                        let mut rest_ctx = ctx.lock().await;
-                        rest_ctx.sensor_channel = Some(svc.id() as u8);
-                    }
+                    let mut sc_lock = sensor_channel.lock().await;
+                    *sc_lock = Some(svc.id() as u8);
                 }
             }
 
@@ -652,7 +649,7 @@ pub async fn proxy<A: Endpoint<A> + 'static>(
     mut rx: Receiver<Packet>,
     mut rxr: Receiver<Packet>,
     config: AppConfig,
-    rest_ctx: Option<Arc<tokio::sync::Mutex<RestContext>>>,
+    sensor_channel: Arc<tokio::sync::Mutex<Option<u8>>>,
 ) -> Result<()> {
     let passthrough = !config.mitm;
     let hex_requested = config.hexdump_level;
@@ -798,7 +795,7 @@ pub async fn proxy<A: Endpoint<A> + 'static>(
                 config.video_in_motion,
                 config.ev,
                 &mut ctx,
-                rest_ctx.clone(),
+                sensor_channel.clone(),
                 config.ev_battery_logger.clone(),
             )
             .await?;
@@ -843,7 +840,7 @@ pub async fn proxy<A: Endpoint<A> + 'static>(
                         config.video_in_motion,
                         config.ev,
                         &mut ctx,
-                        rest_ctx.clone(),
+                        sensor_channel.clone(),
                         config.ev_battery_logger.clone(),
                     )
                     .await?;

src/web.rs

@@ -21,6 +21,7 @@ use std::sync::Arc;
 use tokio::fs;
 use tokio::io::AsyncWriteExt;
 use tokio::sync::mpsc::Sender;
+use tokio::sync::Mutex;
 
 const TEMPLATE: &str = include_str!("../static/index.html");
 const PICO_CSS: &str = include_str!("../static/pico.min.css");
@@ -32,8 +33,8 @@ const NAME: &str = "<i><bright-black> web: </>";
 pub struct AppState {
     pub config: SharedConfig,
     pub config_file: Arc<PathBuf>,
-    pub tx: Option<Sender<Packet>>,
-    pub sensor_channel: Arc<Option<u8>>,
+    pub tx: Arc<Mutex<Option<Sender<Packet>>>>,
+    pub sensor_channel: Arc<Mutex<Option<u8>>>,
 }
 
 pub fn app(state: Arc<AppState>) -> Router {
@@ -70,8 +71,8 @@ pub async fn battery_handler(
 
     info!("{} Received battery level: {}", NAME, data.battery_level);
 
-    if let Some(ch) = *state.sensor_channel {
-        if let Some(tx) = &state.tx {
+    if let Some(ch) = *state.sensor_channel.lock().await {
+        if let Some(tx) = state.tx.lock().await.clone() {
             if let Err(e) = send_ev_data(tx.clone(), data.battery_level, ch, 0, 0.0).await {
                 error!("{} EV model error: {}", NAME, e);
             }