Refactor event listeners initialization#1476
Conversation
… Step 1 This is the first step in a bigger refactor to move the start of event listeners from app container instantiation to app start (jobs creation).
734a23c to
17fb909
Compare
We are moving the execution of the event listener from AppContainer initialization to jobs start. However, in tests we still have to run it manually if we need it.
|
Hi @da2ce7, I've completed the refactor for one of the three packages that have events. I will apply the same refactor to the other two. As I was expecting, there is a problem shutting down the tracker when I wait for the event listener to finish because we don't send a halt message to that task. We can include the join handle in the jobs vector once we implement this issue. We should also improve the shutdown message.
That message is only for axum servers. We should:
It continues to display the message even if all Axum servers are successfully shut down, but the main tracker cannot stop because there are other pending jobs, such as event listeners, in this case. I will open a new issue to address this improvement. I think we should show the job that is preventing the tracker from shutting down if that task does not provide that feature, like the axum server. We should show something like: Every second or so. Maybe we should add this at the main level for all jobs regardless of whether the jobs internally show other messages while shutting down (like axum server). This is a draft, non-tested implementation from ChatGPT 4o: use torrust_tracker_lib::app;
use tokio::time::{sleep, Duration};
use tokio::task::JoinHandle;
use std::sync::{Arc};
use std::sync::atomic::{AtomicBool, Ordering};
#[tokio::main]
async fn main() {
let (_app_container, raw_jobs) = app::run().await;
// Create flags for each job
let mut job_flags = Vec::new();
let mut wrapped_jobs = Vec::new();
for (i, job) in raw_jobs.into_iter().enumerate() {
let running = Arc::new(AtomicBool::new(true));
let running_clone = running.clone();
job_flags.push((format!("Job {}", i + 1), running.clone()));
let handle: JoinHandle<()> = tokio::spawn(async move {
job.await;
running_clone.store(false, Ordering::SeqCst);
});
wrapped_jobs.push(handle);
}
// Spawn the status printer
let status_handle = tokio::spawn({
let job_flags = job_flags.clone();
async move {
loop {
sleep(Duration::from_secs(5)).await;
let still_running: Vec<_> = job_flags
.iter()
.filter(|(_, flag)| flag.load(Ordering::SeqCst))
.collect();
if still_running.is_empty() {
break;
}
println!("!! shutting tracker !!");
for (name, _) in still_running {
println!("Waiting for {name} to finish ...");
}
println!();
}
}
});
// Wait for CTRL+C
tokio::signal::ctrl_c().await.unwrap();
tracing::info!("Torrust tracker shutting down ...");
// Wait for all job handles
for handle in wrapped_jobs {
let _ = handle.await;
}
// Wait for the status task to exit
let _ = status_handle.await;
tracing::info!("Torrust tracker successfully shutdown.");
} |
5773c1b to
e13fb47
Compare
e13fb47 to
07c5858
Compare
|
ACK 6d50a78 |
1 participant
Move the spawning of new tasks for event listeners from app container instantiation to job execution.
This will enable us to easily add more event listeners in the future. Additionally, it enhances the way the tracker handles spawned tasks. We should have control over all JoinHandles for all tasks.
This promotes a cleaner task lifecycle. Some good practices:
Subtasks
setup::factoryfunction.KeepertoHttpTrackerCoreContainer(so we can run the listener later).NOTE: This PR will not include the refactor to allow multiple listeners. See #1385.