-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feature: Add mechanism to synchronize and update services status (Upd…
…ated from #30) (#41) Co-authored-by: danielsanchezq <[email protected]>
- Loading branch information
1 parent
1b83775
commit f5f7ea0
Showing
8 changed files
with
380 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,98 @@ | ||
| // std | ||
| use std::default::Default; | ||
| use std::marker::PhantomData; | ||
| use std::sync::Arc; | ||
| use std::time::Duration; | ||
| // crates | ||
| use crate::services::{ServiceData, ServiceId}; | ||
| use thiserror::Error; | ||
| use tokio::sync::watch; | ||
| // internal | ||
|
|
||
| #[derive(Error, Debug)] | ||
| pub enum ServiceStatusError { | ||
| #[error("service {service_id} is not available")] | ||
| Unavailable { service_id: ServiceId }, | ||
| } | ||
|
|
||
| pub type ServiceStatusResult = Result<StatusWatcher, ServiceStatusError>; | ||
|
|
||
| #[derive(Copy, Clone, Eq, PartialEq, Debug)] | ||
| pub enum ServiceStatus { | ||
| Uninitialized, | ||
| Running, | ||
| Stopped, | ||
| } | ||
|
|
||
| pub struct StatusUpdater(watch::Sender<ServiceStatus>); | ||
|
|
||
| impl StatusUpdater { | ||
| pub fn update(&self, status: ServiceStatus) { | ||
| self.0 | ||
| .send(status) | ||
| .expect("Overwatch always maintain an open watcher, send should always succeed") | ||
| } | ||
| } | ||
|
|
||
| #[derive(Debug, Clone)] | ||
| pub struct StatusWatcher(watch::Receiver<ServiceStatus>); | ||
|
|
||
| impl StatusWatcher { | ||
| pub async fn wait_for( | ||
| &mut self, | ||
| status: ServiceStatus, | ||
| timeout_duration: Option<Duration>, | ||
| ) -> Result<ServiceStatus, ServiceStatus> { | ||
| let current = *self.0.borrow(); | ||
| if status == current { | ||
| return Ok(current); | ||
| } | ||
| let timeout_duration = timeout_duration.unwrap_or_else(|| Duration::from_secs(u64::MAX)); | ||
| tokio::time::timeout(timeout_duration, self.0.wait_for(|s| s == &status)) | ||
| .await | ||
| .map(|r| r.map(|s| *s).map_err(|_| current)) | ||
| .unwrap_or(Err(current)) | ||
| } | ||
| } | ||
|
|
||
| pub struct StatusHandle<S: ServiceData> { | ||
| updater: Arc<StatusUpdater>, | ||
| watcher: StatusWatcher, | ||
| _phantom: PhantomData<S>, | ||
| } | ||
|
|
||
| impl<S: ServiceData> Clone for StatusHandle<S> { | ||
| fn clone(&self) -> Self { | ||
| Self { | ||
| updater: Arc::clone(&self.updater), | ||
| watcher: self.watcher.clone(), | ||
| _phantom: Default::default(), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl<S: ServiceData> StatusHandle<S> { | ||
| pub fn new() -> Self { | ||
| let (updater, watcher) = watch::channel(ServiceStatus::Uninitialized); | ||
| let updater = Arc::new(StatusUpdater(updater)); | ||
| let watcher = StatusWatcher(watcher); | ||
| Self { | ||
| updater, | ||
| watcher, | ||
| _phantom: Default::default(), | ||
| } | ||
| } | ||
| pub fn updater(&self) -> &StatusUpdater { | ||
| &self.updater | ||
| } | ||
|
|
||
| pub fn watcher(&self) -> StatusWatcher { | ||
| self.watcher.clone() | ||
| } | ||
| } | ||
|
|
||
| impl<S: ServiceData> Default for StatusHandle<S> { | ||
| fn default() -> Self { | ||
| Self::new() | ||
| } | ||
| } |
Oops, something went wrong.