openzeppelin_relayer/bootstrap/
initialize_workers.rs

1//! Worker initialization
2//!
3//! Re-exports from the queue backend module where the Apalis-specific worker
4//! logic now lives alongside the Redis backend implementation.
5//!
6//! Also provides `initialize_queue_workers` which consolidates the entire
7//! queue backend lifecycle (creation, worker init) into a single call.
8
9use std::sync::Arc;
10
11use actix_web::web::ThinData;
12use tracing::info;
13
14use crate::{
15    models::DefaultAppState,
16    queues::{QueueBackend, WorkerHandle},
17};
18
19/// Creates the queue backend and initializes all workers in a single step.
20///
21/// This consolidates queue backend creation (from `QUEUE_BACKEND` env var),
22/// worker initialization, and logging into a single bootstrap function,
23/// keeping `main.rs` free of queue implementation details.
24///
25/// # Arguments
26/// * `app_state` - Application state containing the job producer and configuration
27/// * `handle` - Handle to the multi-thread pipeline runtime; workers are spawned
28///   onto it so pipeline work is distributed across its worker threads.
29///
30/// # Returns
31/// Vector of worker handles for all spawned workers
32pub async fn initialize_queue_workers(
33    app_state: ThinData<DefaultAppState>,
34    handle: tokio::runtime::Handle,
35) -> color_eyre::Result<Vec<WorkerHandle>> {
36    let backend = app_state.job_producer.queue_backend();
37
38    let handles = backend
39        .initialize_workers(Arc::new(app_state), handle)
40        .await?;
41
42    info!(
43        backend = %backend.backend_type(),
44        worker_count = handles.len(),
45        "Initialized queue backend workers"
46    );
47
48    Ok(handles)
49}
openzeppelin_relayer/bootstrap/initialize_workers.rs

openzeppelin_relayer/bootstrap/
initialize_workers.rs
