Async worker

src-tauri/Cargo.lock

@@ -207,7 +207,7 @@ dependencies = [
  "cc",
  "cfg-if 1.0.0",
  "constant_time_eq",
- "digest",
+ "digest 0.10.1",
  "rayon",
 ]
 
@@ -217,6 +217,15 @@ version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d8c1fef690941d3e7788d328517591fecc684c084084702d6ff1641e993699a"
 
+[[package]]
+name = "block-buffer"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4152116fd6e9dadb291ae18fc1ec3575ed6d84c29642d97890f4b4a3417297e4"
+dependencies = [
+ "generic-array",
+]
+
 [[package]]
 name = "block-buffer"
 version = "0.10.0"
@@ -528,6 +537,15 @@ dependencies = [
  "objc",
 ]
 
+[[package]]
+name = "cpufeatures"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "95059428f66df56b63431fdb4e1947ed2190586af5c5a8a8b71122bdf5a7f469"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "crc32fast"
 version = "1.3.1"
@@ -702,13 +720,22 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "digest"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3dd60d1080a57a05ab032377049e0591415d2b31afd7028356dbf3cc6dcb066"
+dependencies = [
+ "generic-array",
+]
+
 [[package]]
 name = "digest"
 version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b697d66081d42af4fba142d56918a3cb21dc8eb63372c6b85d14f44fb9c5979b"
 dependencies = [
- "block-buffer",
+ "block-buffer 0.10.0",
  "crypto-common",
  "generic-array",
  "subtle",
@@ -1330,6 +1357,31 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "h2"
+version = "0.3.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9f1f717ddc7b2ba36df7e871fd88db79326551d3d6f1fc406fbfd28b582ff8e"
+dependencies = [
+ "bytes",
+ "fnv",
+ "futures-core",
+ "futures-sink",
+ "futures-util",
+ "http",
+ "indexmap",
+ "slab",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
+[[package]]
+name = "hashbrown"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ab5ef0d4909ef3724cc8cce6ccc8572c5c817592e9285f5464f8e86f8bd3726e"
+
 [[package]]
 name = "heck"
 version = "0.3.3"
@@ -1373,18 +1425,65 @@ dependencies = [
  "itoa 1.0.1",
 ]
 
+[[package]]
+name = "http-body"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ff4f84919677303da5f147645dbea6b1881f368d03ac84e1dc09031ebd7b2c6"
+dependencies = [
+ "bytes",
+ "http",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "http-range"
 version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eee9694f83d9b7c09682fdb32213682939507884e5bcf227be9aff5d644b90dc"
 
+[[package]]
+name = "httparse"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "acd94fdbe1d4ff688b67b04eee2e17bd50995534a61539e45adfefb45e5e5503"
+
+[[package]]
+name = "httpdate"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"
+
 [[package]]
 name = "humantime"
 version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
 
+[[package]]
+name = "hyper"
+version = "0.14.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7ec3e62bdc98a2f0393a5048e4c30ef659440ea6e0e572965103e72bd836f55"
+dependencies = [
+ "bytes",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "h2",
+ "http",
+ "http-body",
+ "httparse",
+ "httpdate",
+ "itoa 0.4.8",
+ "pin-project-lite",
+ "socket2",
+ "tokio",
+ "tower-service",
+ "tracing",
+ "want",
+]
+
 [[package]]
 name = "ico"
 version = "0.1.0"
@@ -1430,6 +1529,16 @@ dependencies = [
  "winapi-util",
 ]
 
+[[package]]
+name = "indexmap"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "282a6247722caba404c065016bbfa522806e51714c34f5dfc3e4a3a46fcb4223"
+dependencies = [
+ "autocfg",
+ "hashbrown",
+]
+
 [[package]]
 name = "infer"
 version = "0.4.0"
@@ -1724,6 +1833,28 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "mio"
+version = "0.7.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8067b404fe97c70829f082dec8bcf4f71225d7eaea1d8645349cb76fa06205cc"
+dependencies = [
+ "libc",
+ "log",
+ "miow",
+ "ntapi",
+ "winapi",
+]
+
+[[package]]
+name = "miow"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
+dependencies = [
+ "winapi",
+]
+
 [[package]]
 name = "native-tls"
 version = "0.2.8"
@@ -1850,6 +1981,15 @@ dependencies = [
  "zvariant_derive",
 ]
 
+[[package]]
+name = "ntapi"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f6bb902e437b6d86e03cce10a7e2af662292c5dfef23b65899ea3ac9354ad44"
+dependencies = [
+ "winapi",
+]
+
 [[package]]
 name = "num-integer"
 version = "0.1.44"
@@ -1946,6 +2086,12 @@ version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "da32515d9f6e6e489d7bc9d84c71b060db7247dc035bbe44eac88cf87486d8d5"
 
+[[package]]
+name = "opaque-debug"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5"
+
 [[package]]
 name = "open"
 version = "2.0.2"
@@ -2847,6 +2993,19 @@ dependencies = [
  "stable_deref_trait",
 ]
 
+[[package]]
+name = "sha-1"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "99cd6713db3cf16b6c84e06321e049a9b9f699826e16096d23bbcc44d15d51a6"
+dependencies = [
+ "block-buffer 0.9.0",
+ "cfg-if 1.0.0",
+ "cpufeatures",
+ "digest 0.9.0",
+ "opaque-debug",
+]
+
 [[package]]
 name = "sharded-slab"
 version = "0.1.4"
@@ -2884,6 +3043,8 @@ version = "0.1.0"
 dependencies = [
  "directories",
  "env_logger",
+ "futures",
+ "hyper",
  "log",
  "palette",
  "rusb",
@@ -2892,6 +3053,9 @@ dependencies = [
  "serialport",
  "tauri",
  "tauri-build",
+ "tokio",
+ "tokio-tungstenite",
+ "tungstenite",
  "vigem-client",
  "winapi",
 ]
@@ -3404,14 +3568,55 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
 [[package]]
 name = "tokio"
-version = "1.15.0"
+version = "1.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbbf1c778ec206785635ce8ad57fe52b3009ae9e0c9f574a728f3049d3e55838"
+checksum = "0c27a64b625de6d309e8c57716ba93021dccf1b3b5c97edd6d3dd2d2135afc0a"
 dependencies = [
  "bytes",
+ "libc",
  "memchr",
+ "mio",
  "num_cpus",
  "pin-project-lite",
+ "tokio-macros",
+ "winapi",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b557f72f448c511a979e2564e55d74e6c4432fc96ff4f6241bc6bded342643b7"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "tokio-tungstenite"
+version = "0.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e80b39df6afcc12cdf752398ade96a6b9e99c903dfdc36e53ad10b9c366bca72"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite",
+]
+
+[[package]]
+name = "tokio-util"
+version = "0.6.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e99e1983e5d376cd8eb4b66604d2e99e79f5bd988c3055891dcd8c9e2604cc0"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "log",
+ "pin-project-lite",
+ "tokio",
 ]
 
 [[package]]
@@ -3423,6 +3628,12 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "tower-service"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "360dfd1d6d30e05fda32ace2c8c70e9c0a9da713275777f5a4dbb8a1893930c6"
+
 [[package]]
 name = "tracing"
 version = "0.1.29"
@@ -3484,6 +3695,31 @@ dependencies = [
  "tracing-log",
 ]
 
+[[package]]
+name = "try-lock"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642"
+
+[[package]]
+name = "tungstenite"
+version = "0.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ad3713a14ae247f22a728a0456a545df14acf3867f905adff84be99e23b3ad1"
+dependencies = [
+ "base64",
+ "byteorder",
+ "bytes",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.8.4",
+ "sha-1",
+ "thiserror",
+ "url",
+ "utf-8",
+]
+
 [[package]]
 name = "typenum"
 version = "1.15.0"
@@ -3607,6 +3843,16 @@ dependencies = [
  "winapi-util",
 ]
 
+[[package]]
+name = "want"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0"
+dependencies = [
+ "log",
+ "try-lock",
+]
+
 [[package]]
 name = "wasi"
 version = "0.9.0+wasi-snapshot-preview1"

src-tauri/Cargo.toml

@@ -15,17 +15,25 @@ build = "src/build.rs"
 tauri-build = { version = "1.0.0-beta.4" }
 
 [dependencies]
+
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
+log = "0.4.14"
+env_logger = "0.9.0"
 tauri = { version = "1.0.0-beta.8", features = ["api-all", "system-tray"] }
+
+directories = "4.0.1"
 rusb = "0.9.0"
 serialport = "4.0.1"
 vigem-client = "0.1.1"
 palette = "0.6.0"
 winapi = "0.3.9"
-directories = "4.0.1"
+
-log = "0.4.14"
+futures = "0.3.19"
-env_logger = "0.9.0"
+tokio = { version="1.16.1", features=["rt-multi-thread","macros"] }
+hyper = { version="0.14.16", features=["server","http1","http2","tcp"] }
+tungstenite = { version="0.16.0", default-features=false }
+tokio-tungstenite = "0.16.1"
 
 [features]
 default = [ "custom-protocol" ]

src-tauri/src/bin/test_async.rs

@@ -0,0 +1,39 @@
+extern crate slidershim;
+
+use std::{io, time::Duration};
+
+use tokio::time::sleep;
+
+// use slidershim::slider_io::worker::{AsyncJob, AsyncJobFut, AsyncJobRecvStop, AsyncWorker};
+
+// struct CounterJob;
+
+// impl AsyncJob for CounterJob {
+//   fn job(self, mut recv_stop: AsyncJobRecvStop) -> AsyncJobFut {
+//     return Box::pin(async move {
+//       let mut x = 0;
+//       loop {
+//         x += 1;
+//         println!("{}", x);
+//         sleep(Duration::from_millis(500)).await;
+//         match recv_stop.try_recv() {
+//           Ok(_) => {
+//             println!("@@@");
+//             break;
+//           }
+//           _ => {}
+//         }
+//       }
+//     });
+//   }
+// }
+
+fn main() {
+  env_logger::Builder::new()
+    .filter_level(log::LevelFilter::Debug)
+    .init();
+
+  // let worker = AsyncWorker::new("counter", CounterJob);
+  let mut input = String::new();
+  let string = io::stdin().read_line(&mut input).unwrap();
+}

src-tauri/src/main.rs

@@ -69,7 +69,7 @@ fn main() {
     .setup(move |app| {
       let app_handle = app.handle();
       let config_clone = Arc::clone(&config);
-      app.listen_global("heartbeat", move |e| {
+      app.listen_global("heartbeat", move |_| {
         let config_handle = config_clone.lock().unwrap();
         info!("Heartbeat received");
         app_handle

src-tauri/src/slider_io/brokenithm.rs

@@ -0,0 +1,40 @@
+use std::{convert::Infallible, net::SocketAddr};
+
+use log::info;
+use tokio::time::sleep;
+
+use hyper::{
+  server::conn::AddrStream,
+  service::{make_service_fn, service_fn},
+  Body, Request, Response, Server,
+};
+
+async fn handle_request(
+  request: Request<Body>,
+  remote_addr: SocketAddr,
+) -> Result<Response<Body>, Infallible> {
+  Ok(Response::new(Body::from(format!(
+    "Hello there connection {}\n",
+    remote_addr
+  ))))
+}
+
+pub async fn brokenithm_server() {
+  let addr = SocketAddr::from(([0, 0, 0, 0], 1666));
+
+  info!("Brokenithm opening on {:?}", addr);
+
+  let make_svc = make_service_fn(|conn: &AddrStream| {
+    let remote_addr = conn.remote_addr();
+    async move {
+      Ok::<_, Infallible>(service_fn(move |request: Request<Body>| {
+        handle_request(request, remote_addr)
+      }))
+    }
+  });
+
+  let server = Server::bind(&addr).serve(make_svc);
+  if let Err(e) = server.await {
+    eprintln!("Server error: {}", e);
+  }
+}

src-tauri/src/slider_io/device.rs

@@ -13,7 +13,7 @@ use crate::slider_io::{
   config::DeviceMode,
   controller_state::{ControllerState, FullState, LedState},
   utils::{Buffer, ShimError},
-  worker::Job,
+  worker::ThreadJob,
 };
 
 type HidReadCallback = fn(&Buffer, &mut ControllerState) -> ();
@@ -212,7 +212,7 @@ impl HidDeviceJob {
 
 const TIMEOUT: Duration = Duration::from_millis(20);
 
-impl Job for HidDeviceJob {
+impl ThreadJob for HidDeviceJob {
   fn setup(&mut self) -> bool {
     match self.setup_impl() {
       Ok(r) => {

src-tauri/src/slider_io/led.rs

@@ -14,7 +14,7 @@ use crate::slider_io::{
   controller_state::{FullState, LedState},
   utils::Buffer,
   voltex::VoltexState,
-  worker::Job,
+  worker::ThreadJob,
 };
 
 pub struct LedJob {
@@ -150,7 +150,7 @@ impl LedJob {
   }
 }
 
-impl Job for LedJob {
+impl ThreadJob for LedJob {
   fn setup(&mut self) -> bool {
     match &self.mode {
       LedMode::Serial { port } => {

src-tauri/src/slider_io/manager.rs

@@ -2,15 +2,15 @@ use log::info;
 
 use crate::slider_io::{
   config::Config, controller_state::FullState, device::HidDeviceJob, led::LedJob,
-  output::OutputJob, worker::Worker,
+  output::OutputJob, worker::ThreadWorker,
 };
 
 pub struct Manager {
   state: FullState,
   config: Config,
-  device_worker: Worker,
+  device_worker: ThreadWorker,
-  output_worker: Worker,
+  output_worker: ThreadWorker,
-  led_worker: Worker,
+  led_worker: ThreadWorker,
 }
 
 impl Manager {
@@ -21,9 +21,12 @@ impl Manager {
     info!("LED config {:?}", config.led_mode);
 
     let state = FullState::new();
-    let device_worker = Worker::new(HidDeviceJob::from_config(&state, &config.device_mode));
+    let device_worker = ThreadWorker::new(
-    let output_worker = Worker::new(OutputJob::new(&state, &config.output_mode));
+      "device",
-    let led_worker = Worker::new(LedJob::new(&state, &config.led_mode));
+      HidDeviceJob::from_config(&state, &config.device_mode),
+    );
+    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,

src-tauri/src/slider_io/mod.rs

@@ -1,10 +1,11 @@
 mod config;
 mod utils;
-mod worker;
+pub mod worker;
 
 mod controller_state;
 mod voltex;
 
+mod brokenithm;
 mod gamepad;
 mod keyboard;
 

src-tauri/src/slider_io/output.rs

@@ -5,7 +5,7 @@ use crate::slider_io::{
   controller_state::FullState,
   gamepad::GamepadOutput,
   keyboard::KeyboardOutput,
-  worker::Job,
+  worker::ThreadJob,
 };
 
 pub trait OutputHandler: Send + Drop {
@@ -38,7 +38,7 @@ impl OutputJob {
   }
 }
 
-impl Job for OutputJob {
+impl ThreadJob for OutputJob {
   fn setup(&mut self) -> bool {
     true
   }

src-tauri/src/slider_io/worker.rs

@@ -1,4 +1,6 @@
 use std::{
+  future::Future,
+  pin::Pin,
   sync::{
     atomic::{AtomicBool, Ordering},
     Arc,
@@ -6,23 +8,35 @@ use std::{
   thread,
 };
 
-pub trait Job: Send {
+use log::info;
+
+use tokio::{
+  runtime::Runtime,
+  sync::oneshot::{self, Receiver},
+  task,
+};
+
+pub trait ThreadJob: Send {
   fn setup(&mut self) -> bool;
   fn tick(&mut self);
   fn teardown(&mut self);
 }
 
-pub struct Worker {
+pub struct ThreadWorker {
+  name: &'static str,
   thread: Option<thread::JoinHandle<()>>,
   stop_signal: Arc<AtomicBool>,
 }
 
-impl Worker {
+impl ThreadWorker {
-  pub fn new<T: 'static + Job>(mut job: T) -> Self {
+  pub fn new<T: 'static + ThreadJob>(name: &'static str, mut job: T) -> Self {
+    info!("Thread worker starting {}", name);
+
     let stop_signal = Arc::new(AtomicBool::new(false));
 
     let stop_signal_clone = Arc::clone(&stop_signal);
     Self {
+      name,
       thread: Some(thread::spawn(move || {
         let setup_res = job.setup();
         stop_signal_clone.store(!setup_res, Ordering::SeqCst);
@@ -33,6 +47,7 @@ impl Worker {
           }
           job.tick();
         }
+        info!("Thread worker stopping internal {}", name);
         job.teardown();
       })),
       stop_signal,
@@ -40,11 +55,74 @@ impl Worker {
   }
 }
 
-impl Drop for Worker {
+impl Drop for ThreadWorker {
   fn drop(&mut self) {
+    info!("Thread worker stopping {}", self.name);
+
     self.stop_signal.store(true, Ordering::SeqCst);
     if self.thread.is_some() {
       self.thread.take().unwrap().join().ok();
     }
   }
 }
+
+pub type AsyncJobRecvStop = oneshot::Receiver<()>;
+pub type AsyncJobFut = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
+pub trait AsyncJob {
+  fn job(self, recv_stop: AsyncJobRecvStop) -> AsyncJobFut;
+}
+
+pub struct AsyncWorker {
+  name: &'static str,
+  runtime: Runtime,
+  task: Option<task::JoinHandle<()>>,
+  stop_signal: Option<oneshot::Sender<()>>,
+}
+
+impl AsyncWorker {
+  pub fn new<T: 'static + AsyncJob + Send>(name: &'static str, job: T) -> 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 fut = job.job(recv_stop);
+      fut.await;
+    });
+
+    AsyncWorker {
+      name,
+      runtime,
+      task: Some(task),
+      stop_signal: Some(send_stop),
+    }
+  }
+}
+
+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();
+      self.runtime.block_on(async move {
+        send_stop.send(()).unwrap();
+      });
+    }
+
+    let name = self.name;
+
+    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);
+      });
+    }
+  }
+}