Frontend polling

2022-02-05 14:16:14 +08:00
parent bfb7f7dc95
commit 8a67094c9a
13 changed files with 394 additions and 103 deletions
@@ -2,7 +2,7 @@ extern crate slidershim;

 use std::io;

-use slidershim::slider_io::{Config, Manager};
+use slidershim::slider_io::{Config, Context};

 fn main() {
  env_logger::Builder::new()
@@ -57,7 +57,7 @@ fn main() {
  // )
  // .unwrap();

-  let manager = Manager::new(config);
+  let manager = Context::new(config);

  let mut input = String::new();
  let string = io::stdin().read_line(&mut input).unwrap();
@@ -18,10 +18,12 @@ use tauri::{
 };

 fn show_window<R: Runtime>(handle: &AppHandle<R>) {
+  handle.emit_all("ackShow", "");
  handle.get_window("main").unwrap().show().ok();
 }

 fn hide_window<R: Runtime>(handle: &AppHandle<R>) {
+  handle.emit_all("ackHide", "");
  handle.get_window("main").unwrap().hide().ok();
 }

@@ -36,14 +38,13 @@ fn main() {
    .init();

  let config = Arc::new(Mutex::new(Some(slider_io::Config::default())));
-  let manager: Arc<Mutex<Option<slider_io::Manager>>> = Arc::new(Mutex::new(None));
+  let manager = Arc::new(Mutex::new(slider_io::Manager::new()));
  {
    let config_handle = config.lock().unwrap();
    let config_handle_ref = config_handle.as_ref().unwrap();
    config_handle_ref.save();
-    // let mut manager_handle = manager.lock().unwrap();
-    // manager_handle.take();
-    // manager_handle.replace(slider_io::Manager::new(config_handle_ref.clone()));
+    let manager_handle = manager.lock().unwrap();
+    manager_handle.update_config(config_handle_ref.clone());
  }

  tauri::Builder::default()
@@ -95,12 +96,9 @@ fn main() {
      // UI ready event
      let app_handle = app.handle();
      let config_clone = Arc::clone(&config);
-      app.listen_global("heartbeat", move |_| {
-        let handle = AsyncHandle::try_current();
-        println!("handle, {:?}", handle);
-
+      app.listen_global("ready", move |_| {
        let config_handle = config_clone.lock().unwrap();
-        info!("Heartbeat received");
+        info!("Start signal received");
        app_handle
          .emit_all(
            "showConfig",
@@ -109,6 +107,23 @@ fn main() {
          .unwrap();
      });

+      // UI update event
+      let app_handle = app.handle();
+      let manager_clone = Arc::clone(&manager);
+      app.listen_global("queryState", move |event| {
+        // app_handle.emit_all("showState", "@@@");
+        let snapshot = {
+          let manager_handle = manager_clone.lock().unwrap();
+          manager_handle.try_get_state().map(|x| x.snapshot())
+        };
+        match snapshot {
+          Some(snapshot) => {
+            app_handle.emit_all("showState", snapshot);
+          }
+          _ => {}
+        }
+      });
+
      // Config set event
      let config_clone = Arc::clone(&config);
      let manager_clone = Arc::clone(&manager);
@@ -121,9 +136,8 @@ fn main() {
          config_handle.replace(new_config);
          let config_handle_ref = config_handle.as_ref().unwrap();
          config_handle_ref.save();
-          let mut manager_handle = manager_clone.lock().unwrap();
-          manager_handle.take();
-          manager_handle.replace(slider_io::Manager::new(config_handle_ref.clone()));
+          let manager_handle = manager_clone.lock().unwrap();
+          manager_handle.update_config(config_handle_ref.clone());
        }
      });

@@ -9,7 +9,7 @@ use hyper::{
 };
 use log::{error, info};
 use path_clean::PathClean;
-use std::{convert::Infallible, future::Future, net::SocketAddr, path::PathBuf};
+use std::{convert::Infallible, env::current_exe, future::Future, net::SocketAddr};
 use tokio::fs::File;
 use tokio_tungstenite::WebSocketStream;
 use tokio_util::codec::{BytesCodec, FramedRead};
@@ -29,15 +29,17 @@ async fn error_response() -> Result<Response<Body>, Infallible> {
 }

 async fn serve_file(path: &str) -> Result<Response<Body>, Infallible> {
-  let mut pb = PathBuf::from("res/www/");
+  let mut pb = current_exe().unwrap();
+  pb.pop();
+  pb.push("res/www");
  pb.push(path);
  pb.clean();

  // println!("CWD {:?}", std::env::current_dir());
-  // println!("Serving file {:?}", pb);

-  match File::open(pb).await {
+  match File::open(&pb).await {
    Ok(f) => {
+      info!("Serving file {:?}", pb);
      let stream = FramedRead::new(f, BytesCodec::new());
      let body = Body::wrap_stream(stream);
      Ok(Response::new(body))
@@ -154,6 +156,7 @@ async fn handle_request(
  request: Request<Body>,
  remote_addr: SocketAddr,
  state: FullState,
+  ground_only: bool,
 ) -> Result<Response<Body>, Infallible> {
  let method = request.method();
  let path = request.uri().path();
@@ -170,7 +173,10 @@ async fn handle_request(
    request.uri().path(),
    request.headers().contains_key(header::UPGRADE),
  ) {
-    ("/", false) | ("/index.html", false) => serve_file("index.html").await,
+    ("/", false) | ("/index.html", false) => match ground_only {
+      false => serve_file("index.html").await,
+      true => serve_file("index-go.html").await,
+    },
    (filename, false) => serve_file(&filename[1..]).await,
    ("/ws", true) => handle_websocket(request, state).await,
    _ => error_response().await,
@@ -179,12 +185,14 @@ async fn handle_request(

 pub struct BrokenithmJob {
  state: FullState,
+  ground_only: bool,
 }

 impl BrokenithmJob {
-  pub fn new(state: &FullState) -> Self {
+  pub fn new(state: &FullState, ground_only: &bool) -> Self {
    Self {
      state: state.clone(),
+      ground_only: *ground_only,
    }
  }
 }
@@ -193,13 +201,14 @@ impl BrokenithmJob {
 impl AsyncJob for BrokenithmJob {
  async fn run<F: Future<Output = ()> + Send>(self, stop_signal: F) {
    let state = self.state.clone();
+    let ground_only = self.ground_only;
    let make_svc = make_service_fn(|conn: &AddrStream| {
      let remote_addr = conn.remote_addr();
      let make_svc_state = state.clone();
      async move {
        Ok::<_, Infallible>(service_fn(move |request: Request<Body>| {
          let svc_state = make_svc_state.clone();
-          handle_request(request, remote_addr, svc_state)
+          handle_request(request, remote_addr, svc_state, ground_only)
        }))
      }
    });
@@ -0,0 +1,64 @@
+use log::info;
+
+use crate::slider_io::{
+  brokenithm::BrokenithmJob,
+  config::{Config, DeviceMode},
+  controller_state::FullState,
+  device::HidDeviceJob,
+  led::LedJob,
+  output::OutputJob,
+  worker::{AsyncWorker, ThreadWorker},
+};
+
+#[allow(dead_code)]
+pub struct Context {
+  state: FullState,
+  config: Config,
+  device_worker: Option<ThreadWorker>,
+  brokenithm_worker: Option<AsyncWorker>,
+  output_worker: ThreadWorker,
+  led_worker: ThreadWorker,
+}
+
+impl Context {
+  pub fn new(config: Config) -> Self {
+    info!("Context creating");
+    info!("Device config {:?}", config.device_mode);
+    info!("Output config {:?}", config.output_mode);
+    info!("LED config {:?}", config.led_mode);
+
+    let state = FullState::new();
+
+    let (device_worker, brokenithm_worker) = match &config.device_mode {
+      DeviceMode::Brokenithm { ground_only } => (
+        None,
+        Some(AsyncWorker::new(
+          "brokenithm",
+          BrokenithmJob::new(&state, ground_only),
+        )),
+      ),
+      other => (
+        Some(ThreadWorker::new(
+          "device",
+          HidDeviceJob::from_config(&state, other),
+        )),
+        None,
+      ),
+    };
+    let output_worker = ThreadWorker::new("output", OutputJob::new(&state, &config.output_mode));
+    let led_worker = ThreadWorker::new("led", LedJob::new(&state, &config.led_mode));
+
+    Self {
+      state,
+      config,
+      device_worker,
+      brokenithm_worker,
+      output_worker,
+      led_worker,
+    }
+  }
+
+  pub fn clone_state(&self) -> FullState {
+    self.state.clone()
+  }
+}
@@ -74,6 +74,21 @@ impl FullState {
  pub fn clone_led(&self) -> Arc<Mutex<LedState>> {
    Arc::clone(&self.led_state)
  }
+
+  pub fn snapshot(&self) -> Vec<u8> {
+    let mut buf: Vec<u8> = vec![];
+    {
+      let controller_state_handle = self.controller_state.lock().unwrap();
+      buf.extend(controller_state_handle.ground_state);
+      buf.extend(controller_state_handle.air_state);
+    };
+    {
+      let led_state_handle = self.led_state.lock().unwrap();
+      buf.extend(led_state_handle.led_state);
+    };
+
+    buf
+  }
 }

 impl Clone for FullState {
@@ -31,8 +31,8 @@ const YUANCON_KB_MAP: [usize; 41] = [
 #[rustfmt::skip]
 const DEEMO_KB_MAP: [usize; 41] = [
  0x41, 0x41, 0x41, 0x41, // A
-  0x35, 0x35, 0x35, 0x35, // S
-  0x4f, 0x4f, 0x4f, 0x4f, // D
+  0x53, 0x53, 0x53, 0x53, // S
+  0x44, 0x44, 0x44, 0x44, // D
  0x46, 0x46, 0x46, 0x46, // F
  0x4a, 0x4a, 0x4a, 0x4a, // J
  0x4b, 0x4b, 0x4b, 0x4b, // K
@@ -1,57 +1,94 @@
 use log::info;
-
-use crate::slider_io::{
-  brokenithm::BrokenithmJob,
-  config::{Config, DeviceMode},
-  controller_state::FullState,
-  device::HidDeviceJob,
-  led::LedJob,
-  output::OutputJob,
-  worker::{AsyncWorker, ThreadWorker},
+use std::{
+  sync::{Arc, Mutex},
+  thread::{self, JoinHandle},
+};
+use tokio::{
+  select,
+  sync::{mpsc, oneshot},
 };

-#[allow(dead_code)]
+use crate::slider_io::{config::Config, context::Context};
+
+use super::controller_state::FullState;
+
 pub struct Manager {
-  state: FullState,
-  config: Config,
-  device_worker: Option<ThreadWorker>,
-  brokenithm_worker: Option<AsyncWorker>,
-  output_worker: ThreadWorker,
-  led_worker: ThreadWorker,
+  state: Arc<Mutex<Option<FullState>>>,
+  join_handle: Option<JoinHandle<()>>,
+  tx_config: mpsc::UnboundedSender<Config>,
+  tx_stop: Option<oneshot::Sender<()>>,
 }

 impl Manager {
-  pub fn new(config: Config) -> Self {
-    info!("Starting manager");
-    info!("Device config {:?}", config.device_mode);
-    info!("Output config {:?}", config.output_mode);
-    info!("LED config {:?}", config.led_mode);
+  pub fn new() -> Self {
+    let state = Arc::new(Mutex::new(None));
+    let (tx_config, mut rx_config) = mpsc::unbounded_channel::<Config>();
+    let (tx_stop, rx_stop) = oneshot::channel::<()>();

-    let state = FullState::new();
+    let context: Arc<Mutex<Option<Context>>> = Arc::new(Mutex::new(None));

-    let (device_worker, brokenithm_worker) = match &config.device_mode {
-      DeviceMode::Brokenithm { .. } => (
-        None,
-        Some(AsyncWorker::new("brokenithm", BrokenithmJob::new(&state))),
-      ),
-      other => (
-        Some(ThreadWorker::new(
-          "device",
-          HidDeviceJob::from_config(&state, other),
-        )),
-        None,
-      ),
-    };
-    let output_worker = ThreadWorker::new("output", OutputJob::new(&state, &config.output_mode));
-    let led_worker = ThreadWorker::new("led", LedJob::new(&state, &config.led_mode));
+    let state_cloned = Arc::clone(&state);
+    let context_cloned = Arc::clone(&context);
+
+    let join_handle = thread::spawn(move || {
+      info!("Manager thread started");
+      let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .enable_all()
+        .build()
+        .unwrap();
+      runtime.block_on(async move {
+        info!("Manager runtime started");
+
+        select! {
+          _ = async {
+            loop {
+              match rx_config.recv().await {
+                Some(config) => {
+                  info!("Rebuilding context");
+                  let mut context_handle = context_cloned.lock().unwrap();
+                  context_handle.take();
+
+                  let new_context = Context::new(config);
+                  let new_state = new_context.clone_state();
+                  context_handle.replace(new_context);
+
+                  let mut state_handle = state_cloned.lock().unwrap();
+                  state_handle.replace(new_state);
+                },
+                None => {
+                  let mut context_handle = context_cloned.lock().unwrap();
+                  context_handle.take();
+                }
+              }
+            }
+          } => {},
+          _ = rx_stop => {}
+        }
+      });
+    });

    Self {
      state,
-      config,
-      device_worker,
-      brokenithm_worker,
-      output_worker,
-      led_worker,
+      join_handle: Some(join_handle),
+      tx_config,
+      tx_stop: Some(tx_stop),
    }
  }
+
+  pub fn update_config(&self, config: Config) {
+    self.tx_config.send(config).unwrap();
+  }
+
+  pub fn try_get_state(&self) -> Option<FullState> {
+    let state_handle = self.state.lock().unwrap();
+    state_handle.as_ref().map(|x| x.clone())
+  }
+}
+
+impl Drop for Manager {
+  fn drop(&mut self) {
+    self.tx_stop.take().unwrap().send(()).unwrap();
+    self.join_handle.take().unwrap().join().unwrap();
+  }
 }
@@ -13,6 +13,7 @@ mod device;
 mod led;
 mod output;

+mod context;
 mod manager;

 pub use config::Config;
@@ -9,7 +9,7 @@ use std::{
  thread,
 };

-use tokio::{runtime::Runtime, sync::oneshot, task};
+use tokio::{sync::oneshot, task};

 pub trait ThreadJob: Send {
  fn setup(&mut self) -> bool;
@@ -68,7 +68,6 @@ pub trait AsyncJob: Send + 'static {

 pub struct AsyncWorker {
  name: &'static str,
-  runtime: Runtime,
  task: Option<task::JoinHandle<()>>,
  stop_signal: Option<oneshot::Sender<()>>,
 }
@@ -81,13 +80,8 @@ impl AsyncWorker {
    info!("Async worker starting {}", name);

    let (send_stop, recv_stop) = oneshot::channel::<()>();
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-      .worker_threads(1)
-      .enable_all()
-      .build()
-      .unwrap();

-    let task = runtime.spawn(async move {
+    let task = tokio::spawn(async move {
      job
        .run(async move {
          recv_stop.await.unwrap();
@@ -97,7 +91,6 @@ impl AsyncWorker {

    AsyncWorker {
      name,
-      runtime,
      task: Some(task),
      stop_signal: Some(send_stop),
    }
@@ -108,20 +101,7 @@ impl Drop for AsyncWorker {
  fn drop(&mut self) {
    info!("Async worker stopping {}", self.name);

-    if self.stop_signal.is_some() {
-      let send_stop = self.stop_signal.take().unwrap();
-      send_stop.send(()).unwrap();
-      // self.runtime.block_on(async move {
-      //   send_stop.send(()).unwrap();
-      // });
-    }
-
-    if self.task.is_some() {
-      // let task = self.task.take().unwrap();
-      // self.runtime.block_on(async move {
-      //   task.await;
-      //   info!("Async worker stopping internal {}", name);
-      // });
-    }
+    self.stop_signal.take().unwrap().send(()).unwrap();
+    self.task.take();
  }
 }