rustlavel_queue/
plugin.rs

1//! The dashboard, attached the way every optional package is:
2//!
3//! ```ignore
4//! App::new().plugin(QueueDashboard::new(queue).watching(["default", "emails"]))
5//! ```
6//!
7//! One explicit line in `main.rs`, no auto-discovery. Registering it also puts
8//! the queue into the application context, so a handler can reach it with
9//! `req.state::<Arc<dyn Queue>>()` and dispatch a job without the application
10//! wiring that up separately.
11//!
12//! JSON rather than HTML: this is the data a Horizon-style dashboard is drawn
13//! from, and rustlavel-view is not a dependency of this crate.
14
15use crate::job::DEFAULT_QUEUE;
16use crate::queue::{Queue, stats};
17use rustlavel_http::plugin::{Plugin, Setup};
18use rustlavel_http::{Request, Response};
19use std::sync::Arc;
20
21/// A read-only view of what the queue is doing.
22pub struct QueueDashboard {
23    queue: Arc<dyn Queue>,
24    path: String,
25    queues: Vec<String>,
26}
27
28impl QueueDashboard {
29    pub fn new(queue: Arc<dyn Queue>) -> Self {
30        QueueDashboard {
31            queue,
32            path: "/queue".to_string(),
33            queues: vec![DEFAULT_QUEUE.to_string()],
34        }
35    }
36
37    /// Serve the dashboard somewhere other than `/queue`.
38    pub fn at(mut self, path: impl Into<String>) -> Self {
39        self.path = path.into();
40        self
41    }
42
43    /// Which queues to report sizes for.
44    ///
45    /// Named explicitly because a driver cannot list its queues without a scan,
46    /// and a dashboard that scans a jobs table on every page load is a dashboard
47    /// that takes production down.
48    pub fn watching<I, S>(mut self, queues: I) -> Self
49    where
50        I: IntoIterator<Item = S>,
51        S: Into<String>,
52    {
53        self.queues = queues.into_iter().map(Into::into).collect();
54        self
55    }
56}
57
58impl Plugin for QueueDashboard {
59    fn name(&self) -> &'static str {
60        "queue"
61    }
62
63    fn register(self: Box<Self>, setup: &mut Setup<'_>) {
64        let QueueDashboard { queue, path, queues } = *self;
65
66        setup.state(Arc::clone(&queue));
67
68        setup.router.get(&path, move |_request: Request| {
69            let queue = Arc::clone(&queue);
70            let queues = queues.clone();
71
72            async move {
73                match stats(queue.as_ref(), &queues).await {
74                    Ok(body) => Response::json(body),
75                    // A dashboard that cannot read the queue should say so,
76                    // not render an empty page that looks like an idle system.
77                    Err(error) => Response::json(rustlavel_core::Json::object([(
78                        "error",
79                        rustlavel_core::Json::from(error.to_string()),
80                    )]))
81                    .with_status(rustlavel_http::Status::SERVICE_UNAVAILABLE),
82                }
83            }
84        });
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use crate::job::QueuedJob;
92    use crate::memory::MemoryQueue;
93    use rustlavel_core::{Config, Context, Json};
94    use rustlavel_http::TestClient;
95    use rustlavel_http::router::Router;
96
97    /// Register the plugin the way an application would, and return a client.
98    async fn dashboard(queue: Arc<dyn Queue>) -> TestClient {
99        let mut router = Router::new();
100        let config = Config::with_defaults();
101        let mut context = Some(Context::builder());
102
103        let mut setup = Setup { router: &mut router, config: &config, context: &mut context };
104        Box::new(QueueDashboard::new(queue).watching(["default", "emails"]))
105            .register(&mut setup);
106
107        let context = context.expect("the builder survives setup").build();
108        TestClient::new(router).with_context(context)
109    }
110
111    #[tokio::test]
112    async fn the_dashboard_reports_queue_sizes_and_failures() {
113        let queue = Arc::new(MemoryQueue::new());
114        queue.push(QueuedJob::new("a", Json::Null)).await.unwrap();
115        queue.push(QueuedJob::new("b", Json::Null).on_queue("emails")).await.unwrap();
116        queue.push(QueuedJob::new("c", Json::Null).on_queue("emails")).await.unwrap();
117
118        let reserved = queue.pop("default").await.unwrap().unwrap();
119        queue.fail(&reserved, "it went wrong").await.unwrap();
120
121        let body = dashboard(queue).await.get("/queue").await.assert_ok().json();
122
123        assert_eq!(body.get("driver").and_then(Json::as_str), Some("memory"));
124        assert_eq!(body.get("queues.default").and_then(Json::as_i64), Some(0));
125        assert_eq!(body.get("queues.emails").and_then(Json::as_i64), Some(2));
126        assert_eq!(body.get("failed.0.name").and_then(Json::as_str), Some("a"));
127        assert_eq!(body.get("failed.0.error").and_then(Json::as_str), Some("it went wrong"));
128    }
129
130    #[tokio::test]
131    async fn the_dashboard_is_empty_but_valid_on_an_idle_system() {
132        let body = dashboard(Arc::new(MemoryQueue::new())).await.get("/queue").await.assert_ok().json();
133
134        assert_eq!(body.get("queues.default").and_then(Json::as_i64), Some(0));
135        assert_eq!(body.get("failed").and_then(Json::as_array).map(<[Json]>::len), Some(0));
136    }
137
138    #[tokio::test]
139    async fn the_dashboard_can_be_mounted_somewhere_else() {
140        let mut router = Router::new();
141        let config = Config::with_defaults();
142        let mut context = Some(Context::builder());
143        let mut setup = Setup { router: &mut router, config: &config, context: &mut context };
144
145        Box::new(QueueDashboard::new(Arc::new(MemoryQueue::new())).at("/admin/jobs"))
146            .register(&mut setup);
147
148        let client = TestClient::new(router).with_context(context.unwrap().build());
149
150        client.get("/admin/jobs").await.assert_ok();
151        client.get("/queue").await.assert_not_found();
152    }
153}
rustlavel_queue/plugin.rs

rustlavel_queue/
plugin.rs
