alemi/upub
alemi/upub
1
0
Fork 1
Code Issues 4 Pull requests Projects Releases Packages 1 Wiki Activity Actions

fix: use mpsc for wake token

Browse source
This commit is contained in:
əlemi 2024-09-19 17:13:53 +02:00
parent 536d8b7618
commit 133b33f8be
Signed by: alemi
GPG key ID: A4895B84D311642C
1 changed files with 12 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
main.rs
View file

@ -167,8 +167,8 @@ async fn init(args: Args, config: upub::Config) {
return; return;
} }
let (tx_wake, rx_wake) = tokio::sync::watch::channel(false); let (tx_wake, rx_wake) = tokio::sync::mpsc::unbounded_channel();
let wake = CancellationToken(rx_wake); let wake = WakeToken(rx_wake);
let ctx = upub::Context::new(db, domain, config.clone(), Some(Box::new(WakerToken(tx_wake)))) let ctx = upub::Context::new(db, domain, config.clone(), Some(Box::new(WakerToken(tx_wake))))
.await.expect("failed creating server context"); .await.expect("failed creating server context");
@ -219,10 +219,18 @@ async fn init(args: Args, config: upub::Config) {
signals_task.await.expect("failed joining signal handler task"); signals_task.await.expect("failed joining signal handler task");
} }
struct WakerToken(tokio::sync::watch::Sender<bool>); struct WakerToken(tokio::sync::mpsc::UnboundedSender<()>);
impl context::WakerToken for WakerToken { impl context::WakerToken for WakerToken {
fn wake(&self) { fn wake(&self) {
self.0.send_replace(true); self.0.send(()).warn_failed("failed waking up workers");
}
}
struct WakeToken(tokio::sync::mpsc::UnboundedReceiver<()>);
impl worker::WakeToken for WakeToken {
async fn wait(&mut self) {
let _ = self.0.recv().await;
} }
} }
@ -235,12 +243,6 @@ impl worker::StopToken for CancellationToken {
} }
} }
impl worker::WakeToken for CancellationToken {
async fn wait(&mut self) {
self.0.changed().await.err_failed("error waiting for waker token");
}
}
impl routes::ShutdownToken for CancellationToken { impl routes::ShutdownToken for CancellationToken {
async fn event(mut self) { async fn event(mut self) {
self.0.changed().await.warn_failed("cancellation token channel closed, stopping..."); self.0.changed().await.warn_failed("cancellation token channel closed, stopping...");