Merge feature/service-status-events into staging (#55)

* Bump main to staging (#7)

* Setup (#1)

 - Add tokio, serde and sqlx crates
 - Setup .gitignore
 - Setup Cargo.toml
 - Setup pipelines

* Enable Dependabot (#3)

Add dependabot.yml with daily cargo updates configured

* Fix staging pipelines (#4)

Fix staging pipelines not triggering on pull request

* Implement Cargo caching (#5)

Add Swatinem's Rust-Cache@v2 action to build, test and deploy pipelines

* Improve README.md (#6)

 - Add deployment badges
 - Add collaborating info (board + issues)

* Clonable Status

 - Made Status clonable by not using BoxedErrors anymore but Strings for holding the error information
 - Add get_status() to Service
 - Made status property of Service private

* Adapt Service Manager to new Status enum

* Events

 - Implement Event<T>
 - Add status_changed Event to ServiceInfo

* Event improvements

 - Add name attribute to Event<T>
 - Unify Channel and Closure subscribers by using an Enum
 - Propagate errors when dispatching events
 - Add error log when errors occur while dispatching events
 - Subscribers are now removed from an event when they run into an error while dispatching

* Event improvements

Make the removal of Event subscribers on error optional

* Slight refactors

 - Refactors in service_manager.rs
 - Refactors in watchdog.rs

* WIP: Idk lol

I made these changes many months ago. Reviewed them for like half an hour, looks good. I know what I was working on. Will continue now :)

* add: allow clippy::multiple_bound_locations for service module

* add: observables

* refactor: use Mutex instead of RwLock everywhere

* refactor: make remove_on_error work on per-subscriber basis

* refactor: make subscribers identifiable

* refactor: move dispatch logic to Subscriber

* add: AsyncClosure Callback type

* WIP: EventRepeater

* add: EventRepeater

* refactor: event subscribe method names

* refactor: move subscription into own module

* add: AsRef<Event<T>>

* add: UUID, PartialEq/Eq, unsubscribe()

* add: event_repeater detach(), close()

* add: attach/deattach EventRepeater on start/stop of service

* add: service runtime failure handling

* fix: bump version to 0.2.1

Author: Kitt3120

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lum"
-version = "0.1.0"
+version = "0.2.1"
 edition = "2021"
 description = "Lum Discord Bot"
 license= "MIT"
@@ -23,3 +23,4 @@ serenity = { version = "0.12.0", default-features=false, features = ["builder",
 sqlx = { version = "0.8.0", features = ["runtime-tokio", "any", "postgres", "mysql", "sqlite", "tls-native-tls", "migrate", "macros", "uuid", "chrono", "json"] }
 thiserror = "1.0.52"
 tokio = { version = "1.35.1", features = ["full"] }
+uuid = { version = "1.10.0", features = ["fast-rng", "macro-diagnostics", "v4"] }

README.md

@@ -14,4 +14,4 @@ Beta: [![Deploy](https://github.com/Kitt3120/lum/actions/workflows/deploy_prerel
 
 ## Collaborating
 
-Checkout out [Milestones](https://github.com/Kitt3120/lum/milestones), [Board](https://github.com/users/Kitt3120/projects/3), and [Issues](https://github.com/Kitt3120/lum/issues)
+Check out [Milestones](https://github.com/Kitt3120/lum/milestones), [Board](https://github.com/users/Kitt3120/projects/3), and [Issues](https://github.com/Kitt3120/lum/issues)

build.rs

@@ -2,4 +2,4 @@
 fn main() {
     // trigger recompilation when a new migration is added
     println!("cargo:rerun-if-changed=migrations");
-}
+}

src/bot.rs

@@ -1,8 +1,27 @@
-use std::sync::Arc;
+use core::fmt;
+use std::{fmt::Display, sync::Arc};
 
-use tokio::sync::RwLock;
+use log::error;
+use tokio::{signal, sync::Mutex};
 
-use crate::service::{PinnedBoxedFuture, Service, ServiceManager, ServiceManagerBuilder};
+use crate::service::{
+    types::LifetimedPinnedBoxedFuture, OverallStatus, Service, ServiceManager, ServiceManagerBuilder,
+};
+
+#[derive(Debug, Clone, Copy)]
+pub enum ExitReason {
+    SIGINT,
+    EssentialServiceFailed,
+}
+
+impl Display for ExitReason {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::SIGINT => write!(f, "SIGINT"),
+            Self::EssentialServiceFailed => write!(f, "Essential Service Failed"),
+        }
+    }
+}
 
 pub struct BotBuilder {
     name: String,
@@ -17,13 +36,13 @@ impl BotBuilder {
         }
     }
 
-    pub async fn with_service(mut self, service: Arc<RwLock<dyn Service>>) -> Self {
+    pub async fn with_service(mut self, service: Arc<Mutex<dyn Service>>) -> Self {
         self.service_manager = self.service_manager.with_service(service).await; // The ServiceManagerBuilder itself will warn when adding a service multiple times
 
         self
     }
 
-    pub async fn with_services(mut self, services: Vec<Arc<RwLock<dyn Service>>>) -> Self {
+    pub async fn with_services(mut self, services: Vec<Arc<Mutex<dyn Service>>>) -> Self {
         for service in services {
             self.service_manager = self.service_manager.with_service(service).await;
         }
@@ -50,18 +69,56 @@ impl Bot {
     }
 
     //TODO: When Rust allows async trait methods to be object-safe, refactor this to use async instead of returning a future
-    pub fn start(&mut self) -> PinnedBoxedFuture<'_, ()> {
+    pub fn start(&mut self) -> LifetimedPinnedBoxedFuture<'_, ()> {
         Box::pin(async move {
             self.service_manager.start_services().await;
             //TODO: Potential for further initialization here, like modules
         })
     }
 
     //TODO: When Rust allows async trait methods to be object-safe, refactor this to use async instead of returning a future
-    pub fn stop(&mut self) -> PinnedBoxedFuture<'_, ()> {
+    pub fn stop(&mut self) -> LifetimedPinnedBoxedFuture<'_, ()> {
         Box::pin(async move {
             self.service_manager.stop_services().await;
             //TODO: Potential for further deinitialization here, like modules
         })
     }
+
+    pub async fn join(&self) -> ExitReason {
+        let name_clone = self.name.clone();
+        let signal_task = tokio::spawn(async move {
+            let name = name_clone;
+
+            let result = signal::ctrl_c().await;
+            if let Err(error) = result {
+                error!(
+                    "Error receiving SIGINT: {}. {} will exit ungracefully immediately to prevent undefined behavior.",
+                    error, name
+                );
+                panic!("Error receiving SIGINT: {}", error);
+            }
+        });
+
+        let service_manager_clone = self.service_manager.clone();
+        let mut receiver = self
+            .service_manager
+            .on_status_change
+            .event
+            .subscribe_channel("t", 2, true, true)
+            .await;
+        let status_task = tokio::spawn(async move {
+            let service_manager = service_manager_clone;
+            while (receiver.receiver.recv().await).is_some() {
+                let overall_status = service_manager.overall_status().await;
+                if overall_status == OverallStatus::Unhealthy {
+                    return;
+                }
+            }
+        });
+
+        tokio::select! {
+            _ = signal_task => ExitReason::SIGINT,
+            _ = status_task => ExitReason::EssentialServiceFailed,
+        }
+    }
 }

src/event.rs

@@ -0,0 +1,13 @@
+pub mod arc_observable;
+pub mod event;
+pub mod event_repeater;
+pub mod observable;
+pub mod subscriber;
+pub mod subscription;
+
+pub use arc_observable::ArcObservable;
+pub use event::Event;
+pub use event_repeater::EventRepeater;
+pub use observable::{Observable, ObservableResult};
+pub use subscriber::{Callback, DispatchError, Subscriber};
+pub use subscription::{ReceiverSubscription, Subscription};

src/event/arc_observable.rs

@@ -0,0 +1,60 @@
+use std::{
+    hash::{DefaultHasher, Hash, Hasher},
+    sync::Arc,
+};
+
+use tokio::sync::Mutex;
+
+use super::{Event, ObservableResult};
+
+#[derive(Debug)]
+pub struct ArcObservable<T>
+where
+    T: Send + 'static + Hash,
+{
+    value: Arc<Mutex<T>>,
+    on_change: Event<Mutex<T>>,
+}
+
+impl<T> ArcObservable<T>
+where
+    T: Send + 'static + Hash,
+{
+    pub fn new(value: T, event_name: impl Into<String>) -> Self {
+        Self {
+            value: Arc::new(Mutex::new(value)),
+            on_change: Event::new(event_name),
+        }
+    }
+
+    pub async fn get(&self) -> Arc<Mutex<T>> {
+        Arc::clone(&self.value)
+    }
+
+    pub async fn set(&self, value: T) -> ObservableResult<Mutex<T>> {
+        let mut lock = self.value.lock().await;
+
+        let mut hasher = DefaultHasher::new();
+        (*lock).hash(&mut hasher);
+        let current_value = hasher.finish();
+
+        let mut hasher = DefaultHasher::new();
+        value.hash(&mut hasher);
+        let new_value = hasher.finish();
+
+        if current_value == new_value {
+            return ObservableResult::Unchanged;
+        }
+
+        *lock = value;
+        drop(lock);
+
+        let value = Arc::clone(&self.value);
+        let dispatch_result = self.on_change.dispatch(value).await;
+
+        match dispatch_result {
+            Ok(_) => ObservableResult::Changed(Ok(())),
+            Err(errors) => ObservableResult::Changed(Err(errors)),
+        }
+    }
+}