pipedash_core/application/
mod.rs

1pub mod fetch_scheduler;
2pub mod services;
3#[cfg(test)]
4pub(crate) mod test_support;
5
6use std::sync::Arc;
7
8pub use fetch_scheduler::{
9    AttemptHandle,
10    BackoffPolicy,
11    ClaimKind,
12    FetchClaim,
13    FetchResult,
14    FetchScheduler,
15    ForcedClaim,
16    Generation,
17    Outcome,
18    ScheduleView,
19};
20pub use services::health_service::{
21    HealthHandle,
22    HealthInput,
23    HealthReceivers,
24    HealthSnapshot,
25    ProviderHealthInput,
26    ProviderHealthService,
27    ProviderHealthState,
28};
29pub use services::metrics_service::MetricsService;
30pub use services::pipeline_service::PipelineService;
31pub use services::provider_service::{
32    ProviderLease,
33    ProviderService,
34};
35use tokio::sync::watch;
36
37use crate::event::EventBus;
38use crate::infrastructure::database::Repository;
39use crate::infrastructure::{
40    ConfigState,
41    HttpClientManager,
42};
43
44mod refresh_manager;
45pub use refresh_manager::{
46    RefreshCommand,
47    RefreshDispatcher,
48    RefreshManager,
49    RefreshMode,
50};
51
52pub struct Services {
53    pub provider_service: Arc<ProviderService>,
54    pub pipeline_service: Arc<PipelineService>,
55    pub refresh_manager: Arc<RefreshManager>,
56    pub health: HealthHandle,
57    pub health_service: Arc<ProviderHealthService>,
58}
59
60impl Services {
61    pub fn new(
62        repository: Arc<Repository>, http_client_manager: Arc<HttpClientManager>,
63        metrics_service: Option<Arc<MetricsService>>, event_bus: Arc<dyn EventBus>,
64        config_state: Arc<ConfigState>,
65    ) -> Self {
66        let (health, health_receivers) = HealthHandle::channel();
67        let (dispatcher, commands) = RefreshDispatcher::channel();
68        let (mode_tx, mode_rx) = watch::channel(RefreshMode::Active);
69        config_state.attach_health(health.clone());
70
71        let provider_service = Arc::new(ProviderService::new(
72            Arc::clone(&repository),
73            Arc::clone(&http_client_manager),
74            Arc::clone(&event_bus),
75            Arc::new(FetchScheduler::default()),
76            health.clone(),
77            dispatcher,
78        ));
79        let health_service = Arc::new(ProviderHealthService::new(
80            health_receivers,
81            Arc::clone(&provider_service),
82            http_client_manager,
83            Arc::clone(&event_bus),
84            config_state,
85            mode_tx.subscribe(),
86        ));
87        let pipeline_service = Arc::new(PipelineService::new(
88            repository,
89            Arc::clone(&provider_service),
90            metrics_service.clone(),
91            event_bus,
92            mode_rx,
93        ));
94        let refresh_manager = Arc::new(RefreshManager::new(
95            Arc::clone(&pipeline_service),
96            metrics_service,
97            mode_tx,
98            commands,
99        ));
100
101        Self {
102            provider_service,
103            pipeline_service,
104            refresh_manager,
105            health,
106            health_service,
107        }
108    }
109}
pipedash_core/application/mod.rs

pipedash_core/application/
mod.rs
