e034b46
use std::path::PathBuf;
use std::sync::Arc;
use config::email::AppEmailSettings;
use config::queue::AppQueueSettings;
use queue::{Queue, QueueBuilder};
use tokio::signal;
use tracing::{info, warn};
use crate::error::EmailError;
use crate::service::{EmailService, EmailServiceBuilder, EmailSubscribers};
#[derive(Debug, Default, Clone)]
pub struct WorkerOptions {
pub template_dir: Option<PathBuf>,
}
pub struct RunningWorker {
pub queue: Arc<Queue>,
pub service: EmailService,
pub subscribers: EmailSubscribers,
}
pub async fn start_worker(
email_cfg: AppEmailSettings,
queue_cfg: AppQueueSettings,
opts: WorkerOptions,
) -> Result<RunningWorker, EmailError> {
let queue = QueueBuilder::from_config(queue_cfg)
.build()
.await?
.ok_or(EmailError::Config("queue disabled at startup".to_string()))?;
let mut builder = EmailServiceBuilder::new(email_cfg, Arc::new(queue.clone()));
if let Some(dir) = opts.template_dir {
builder = builder.with_template_dir(dir);
}
let service = builder.build().await?.ok_or(EmailError::Config(
"email service disabled at startup".to_string(),
))?;
let subscribers = service.start_subscribers()?;
info!("email worker subscribers started");
Ok(RunningWorker {
queue: Arc::new(queue),
service,
subscribers,
})
}
pub async fn wait_for_shutdown() {
let ctrl_c = async {
if let Err(e) = signal::ctrl_c().await {
warn!(error = %e, "failed to install ctrl-c handler");
}
};
#[cfg(unix)]
let term = async {
match signal::unix::signal(signal::unix::SignalKind::terminate()) {
Ok(mut s) => {
s.recv().await;
}
Err(e) => {
warn!(error = %e, "failed to install SIGTERM handler");
std::future::pending::<()>().await;
}
}
};
#[cfg(not(unix))]
let term = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => info!("ctrl-c received, beginning shutdown"),
_ = term => info!("SIGTERM received, beginning shutdown"),
}
}
pub async fn run_until_shutdown(
email_cfg: AppEmailSettings,
queue_cfg: AppQueueSettings,
opts: WorkerOptions,
) -> Result<(), EmailError> {
let running = start_worker(email_cfg, queue_cfg, opts).await?;
wait_for_shutdown().await;
running.subscribers.shutdown().await?;
running.service.shutdown().await?;
info!("email worker shut down cleanly");
Ok(())
}