roadster 0.8.1

A "Batteries Included" web framework for rust designed to get you moving fast.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

//! Background task queue service backed by Redis using [rusty-sidekiq](https://docs.rs/rusty-sidekiq).
//! The [`SidekiqWorkerService`] is a simple wrapper around a [`SidekiqProcessor`].

use crate::app::context::AppContext;
use crate::error::RoadsterResult;
use crate::service::Service;
use crate::worker::backend::sidekiq::processor::SidekiqProcessor;
use async_trait::async_trait;
use axum_core::extract::FromRef;
use tokio_util::sync::CancellationToken;
use tracing::{debug, instrument};

pub(crate) const NAME: &str = "worker-sidekiq";

#[derive(bon::Builder)]
#[non_exhaustive]
pub struct SidekiqWorkerService<S>
where
    S: Clone + Send + Sync + 'static,
    AppContext: FromRef<S>,
{
    pub(crate) processor: SidekiqProcessor<S>,
}

#[async_trait]
impl<S> Service<S> for SidekiqWorkerService<S>
where
    S: Clone + Send + Sync + 'static,
    AppContext: FromRef<S>,
{
    fn name(&self) -> String {
        NAME.to_string()
    }

    fn enabled(&self, state: &S) -> bool {
        enabled(&AppContext::from_ref(state), &self.processor)
    }

    #[instrument(skip_all)]
    async fn before_run(&self, state: &S) -> RoadsterResult<()> {
        self.processor.before_run(state).await?;
        Ok(())
    }

    async fn run(
        self: Box<Self>,
        state: &S,
        cancel_token: CancellationToken,
    ) -> RoadsterResult<()> {
        self.processor.run(state, cancel_token).await;
        Ok(())
    }
}

pub(crate) fn enabled<S>(context: &AppContext, processor: &SidekiqProcessor<S>) -> bool
where
    S: Clone + Send + Sync + 'static,
    AppContext: FromRef<S>,
{
    let config = &context.config().service.worker.sidekiq;
    if !config.common.enabled(context) {
        debug!("Sidekiq is not enabled in the config.");
        return false;
    }

    let dedicated_workers: u64 = config
        .custom
        .common
        .queue_config
        .values()
        .map(|config| u64::from(config.num_workers.unwrap_or_default()))
        .sum();

    if config.custom.common.num_workers == 0 && dedicated_workers == 0 {
        debug!("Sidekiq configured with 0 worker tasks.");
        return false;
    }

    let queues_empty = if let Some(queues) = config.custom.common.queues.as_ref() {
        queues.is_empty()
    } else {
        processor.inner.queues.is_empty()
    };

    if queues_empty && dedicated_workers == 0 {
        debug!("Sidekiq configured with 0 worker queues.");
        return false;
    }

    if context.redis_fetch().is_none() {
        debug!("No 'redis-fetch' pool connections available.");
        return false;
    }

    true
}