add background service scheduler

Cargo.lock

@@ -85,6 +85,7 @@ version = "2.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ea16c295198e958ef31930a6ef37d0fb64e9ca3b6116e6b93a8bdae96ee1000"
 dependencies = [
+ "actix-macros",
  "futures-core",
  "tokio",
 ]
@@ -231,6 +232,17 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "async-trait"
+version = "0.1.59"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "31e6e93155431f3931513b243d371981bb2770112b370c82745a1d19d2f99364"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "atoi"
 version = "1.0.0"
@@ -1024,7 +1036,9 @@ dependencies = [
 name = "nyanoblog"
 version = "0.1.0"
 dependencies = [
+ "actix-rt",
  "actix-web",
+ "async-trait",
  "chrono",
  "dotenvy",
  "log",
@@ -1033,6 +1047,7 @@ dependencies = [
  "serde_json",
  "sqlx",
  "thiserror",
+ "tokio",
 ]
 
 [[package]]
@@ -1620,9 +1635,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
 [[package]]
 name = "tokio"
-version = "1.22.0"
+version = "1.23.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d76ce4a75fb488c605c54bf610f221cea8b0dafb53333c1a67e8ee199dcd2ae3"
+checksum = "eab6d665857cc6ca78d6e80303a02cea7a7851e85dfbd77cbdc09bd129f1ef46"
 dependencies = [
  "autocfg",
  "bytes",
@@ -1634,7 +1649,7 @@ dependencies = [
  "pin-project-lite",
  "signal-hook-registry",
  "socket2",
- "winapi",
+ "windows-sys",
 ]
 
 [[package]]

Cargo.toml

@@ -4,7 +4,9 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+actix-rt = "2.7"
 actix-web = "4"
+async-trait = "0.1.59"
 chrono = { version = "0.4", features = [ "alloc", "clock", "serde" ] }
 dotenvy = "0.15.6"
 log = "0.4"
@@ -13,3 +15,4 @@ serde = "1.0"
 serde_json = "1.0"
 sqlx = { version = "0.6", features = [ "chrono", "runtime-actix-rustls", "postgres" ] }
 thiserror = "1.0"
+tokio = "1.23.0"

src/main.rs

@@ -8,6 +8,7 @@ mod data;
 mod model;
 mod repo;
 mod route;
+mod service;
 mod state;
 use crate::core::*;
 use conf::Config;

src/service/mod.rs

@@ -0,0 +1,47 @@
+use actix_rt::Arbiter;
+use async_trait::async_trait;
+use sqlx::Execute;
+use std::{cell::RefCell, fmt, sync::Mutex};
+use tokio::sync::mpsc::{channel, Receiver, Sender};
+
+use crate::core::*;
+use crate::state::AppState;
+
+#[async_trait]
+pub trait Service: Send + Sync + fmt::Debug {
+    async fn perform(&self) -> Result<()>;
+    fn name(&self) -> &'static str;
+}
+
+pub struct Scheduler {
+    arbiter: Arbiter,
+    sender: Sender<Box<dyn Service>>,
+}
+
+impl Scheduler {
+    pub fn new() -> Scheduler {
+        let (tx, rx) = channel(32);
+        let sched = Scheduler {
+            arbiter: Arbiter::new(),
+            sender: tx,
+        };
+
+        if !sched.arbiter.spawn(worker(rx)) {
+            panic!("Service scheduler initialization failed");
+        }
+        sched
+    }
+
+    pub async fn schedule(&self, service: Box<dyn Service>) {
+        self.sender.send(service).await.expect("Service send failed");
+    }
+}
+
+async fn worker(mut receiver: Receiver<Box<dyn Service>>) {
+    while let Some(service) = receiver.recv().await {
+        debug!(target: "service", "Starting service {}", service.name());
+        if let Err(e) = service.perform().await {
+            error!(target: "service", "Service {} failed: {}", service.name(), e);
+        }
+    }
+}

src/state.rs

@@ -4,10 +4,12 @@ use std::sync::Arc;
 
 use crate::conf::Config;
 use crate::repo::AppRepo;
+use crate::service::Scheduler;
 
 pub struct State {
     pub config: Config,
     pub repo: AppRepo,
+    pub sched: Scheduler,
 }
 
 pub type AppStateRaw = Arc<State>;
@@ -17,5 +19,6 @@ pub fn new(config: Config, db_pool: PgPool) -> AppState {
     web::Data::new(Arc::new(State {
         config,
         repo: AppRepo::new(db_pool),
+        sched: Scheduler::new(),
     }))
 }