apalis_sqlite/queries/
list_workers.rs

1use apalis_core::backend::{BackendExt, ListWorkers, RunningWorker};
2use futures::TryFutureExt;
3use ulid::Ulid;
4
5use crate::{CompactType, SqliteContext, SqliteStorage};
6
7struct Worker {
8    id: String,
9    worker_type: String,
10    storage_name: String,
11    layers: Option<String>,
12    last_seen: i64,
13    started_at: Option<i64>,
14}
15
16impl<Args: Sync, D, F> ListWorkers for SqliteStorage<Args, D, F>
17where
18    Self: BackendExt<
19            Context = SqliteContext,
20            Compact = CompactType,
21            IdType = Ulid,
22            Error = sqlx::Error,
23        >,
24{
25    fn list_workers(
26        &self,
27        queue: &str,
28    ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send {
29        let queue = queue.to_owned();
30        let pool = self.pool.clone();
31        let limit = 100;
32        let offset = 0;
33        async move {
34            let workers = sqlx::query_file_as!(
35                Worker,
36                "queries/backend/list_workers.sql",
37                queue,
38                limit,
39                offset
40            )
41            .fetch_all(&pool)
42            .map_ok(|w| {
43                w.into_iter()
44                    .map(|w| RunningWorker {
45                        id: w.id,
46                        backend: w.storage_name,
47                        started_at: w.started_at.unwrap_or_default() as u64,
48                        last_heartbeat: w.last_seen as u64,
49                        layers: w.layers.unwrap_or_default(),
50                        queue: w.worker_type,
51                    })
52                    .collect()
53            })
54            .await?;
55            Ok(workers)
56        }
57    }
58
59    fn list_all_workers(
60        &self,
61    ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send {
62        let pool = self.pool.clone();
63        let limit = 100;
64        let offset = 0;
65        async move {
66            let workers = sqlx::query_file_as!(
67                Worker,
68                "queries/backend/list_all_workers.sql",
69                limit,
70                offset
71            )
72            .fetch_all(&pool)
73            .map_ok(|w| {
74                w.into_iter()
75                    .map(|w| RunningWorker {
76                        id: w.id,
77                        backend: w.storage_name,
78                        started_at: w.started_at.unwrap_or_default() as u64,
79                        last_heartbeat: w.last_seen as u64,
80                        layers: w.layers.unwrap_or_default(),
81                        queue: w.worker_type,
82                    })
83                    .collect()
84            })
85            .await?;
86            Ok(workers)
87        }
88    }
89}
apalis_sqlite/queries/list_workers.rs

apalis_sqlite/queries/
list_workers.rs
