149bec6
use std::path::PathBuf;
use std::time::Duration;
use tracing::{debug, info, warn};
use crate::error::EmailError;
use crate::service::{EmailService, EmailServiceBuilder};
use crate::subscriber::EmailSubscribers;
#[derive(Default)]
pub struct WorkerOptions {
pub template_dir: Option<PathBuf>,
pub shutdown_timeout: Option<Duration>,
}
pub struct RunningWorker {
pub service: EmailService,
pub subscribers: EmailSubscribers,
}
pub async fn start_worker(
email_cfg: config::email::EmailConfig,
queue_cfg: config::email_queue::EmailQueueConfig,
redis_url: String,
opts: WorkerOptions,
) -> Result<RunningWorker, EmailError> {
let _ = opts.template_dir;
let enabled = email_cfg.enabled();
info!(
smtp_enabled = enabled,
queue_enabled = queue_cfg.enabled(),
"building email service"
);
let builder = EmailServiceBuilder::new(email_cfg)
.with_queue(redis_url, queue_cfg)
.register_default_templates()
.await;
let service = builder.build().await?;
let subscribers = service.start_subscribers().await?;
info!(
smtp_enabled = service.enabled(),
subscribers_started = service.queue().is_some_and(|q| q.enabled()),
"email worker ready"
);
Ok(RunningWorker {
service,
subscribers,
})
}
pub async fn run_until_shutdown(
email_cfg: config::email::EmailConfig,
queue_cfg: config::email_queue::EmailQueueConfig,
redis_url: String,
opts: WorkerOptions,
) -> Result<(), EmailError> {
let worker = start_worker(email_cfg, queue_cfg, redis_url, opts).await?;
if !worker.service.enabled() {
warn!("email service is disabled at startup; exiting immediately");
return Ok(());
}
info!("email worker started; waiting for shutdown signal");
debug!("entering wait_for_shutdown");
wait_for_shutdown().await;
let timeout = Duration::from_secs(5);
debug!(
timeout_secs = timeout.as_secs(),
"calling subscribers.shutdown"
);
worker.subscribers.shutdown(timeout).await?;
info!("email worker shut down");
Ok(())
}
pub async fn wait_for_shutdown() {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut sigterm = match signal(SignalKind::terminate()) {
Ok(s) => s,
Err(err) => {
warn!(error = %err, "failed to install SIGTERM handler");
let _ = tokio::signal::ctrl_c().await;
return;
}
};
tokio::select! {
_ = tokio::signal::ctrl_c() => {
info!("received SIGINT");
}
_ = sigterm.recv() => {
info!("received SIGTERM");
}
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
info!("received ctrl-c");
}
}