pipedash_core/application/
mod.rs

1mod config_watcher;
2pub mod fetch_scheduler;
3pub mod services;
4#[cfg(test)]
5pub(crate) mod test_support;
6
7use std::sync::Arc;
8
9pub use config_watcher::ConfigWatcher;
10pub use fetch_scheduler::{
11    AttemptHandle,
12    BackoffPolicy,
13    ClaimKind,
14    FetchClaim,
15    FetchResult,
16    FetchScheduler,
17    ForcedClaim,
18    Generation,
19    Outcome,
20    ScheduleView,
21};
22pub use services::health_service::{
23    refresh_notifications,
24    HealthHandle,
25    HealthInput,
26    HealthReceivers,
27    HealthSnapshot,
28    LocalNotification,
29    NotificationState,
30    ProviderHealthInput,
31    ProviderHealthService,
32    ProviderHealthState,
33};
34pub use services::log_service::{
35    LogExport,
36    LogService,
37};
38pub use services::metrics_service::MetricsService;
39pub use services::pipeline_service::{
40    PipelineService,
41    RunWatch,
42    WatchedRun,
43};
44pub use services::pipeline_view_service::PipelineViewService;
45pub use services::provider_service::{
46    ProviderLease,
47    ProviderService,
48};
49use tokio::sync::watch;
50
51use crate::event::EventBus;
52use crate::infrastructure::database::Repository;
53use crate::infrastructure::run_logs::LogStore;
54use crate::infrastructure::{
55    ConfigState,
56    HttpClientManager,
57};
58
59mod refresh_manager;
60pub use refresh_manager::{
61    RefreshCommand,
62    RefreshDispatcher,
63    RefreshManager,
64    RefreshMode,
65};
66
67pub struct Services {
68    pub provider_service: Arc<ProviderService>,
69    pub pipeline_service: Arc<PipelineService>,
70    pub refresh_manager: Arc<RefreshManager>,
71    pub health: HealthHandle,
72    pub health_service: Arc<ProviderHealthService>,
73    pub log_service: Arc<LogService>,
74    pub pipeline_view_service: Arc<PipelineViewService>,
75}
76
77impl Services {
78    pub fn new(
79        repository: Arc<Repository>, http_client_manager: Arc<HttpClientManager>,
80        metrics_service: Option<Arc<MetricsService>>, event_bus: Arc<dyn EventBus>,
81        config_state: Arc<ConfigState>, log_store: Arc<LogStore>,
82    ) -> Self {
83        let (health, health_receivers) = HealthHandle::channel();
84        let (dispatcher, commands) = RefreshDispatcher::channel();
85        let (mode_tx, mode_rx) = watch::channel(RefreshMode::Active);
86        config_state.attach_health(health.clone());
87        let view_repository = Arc::clone(&repository);
88        let view_config_state = Arc::clone(&config_state);
89        let view_event_bus = Arc::clone(&event_bus);
90
91        let provider_service = Arc::new(ProviderService::new(
92            Arc::clone(&repository),
93            Arc::clone(&http_client_manager),
94            Arc::clone(&event_bus),
95            Arc::new(FetchScheduler::default()),
96            health.clone(),
97            dispatcher,
98            Arc::clone(&config_state),
99            Arc::clone(&log_store),
100        ));
101        let health_service = Arc::new(ProviderHealthService::new(
102            health_receivers,
103            Arc::clone(&provider_service),
104            http_client_manager,
105            Arc::clone(&event_bus),
106            config_state,
107            mode_tx.subscribe(),
108        ));
109        let log_service = Arc::new(LogService::new(
110            log_store,
111            Arc::clone(&provider_service),
112            Arc::clone(&repository),
113            Arc::clone(&event_bus),
114        ));
115        let pipeline_service = Arc::new(PipelineService::new(
116            repository,
117            Arc::clone(&provider_service),
118            metrics_service.clone(),
119            event_bus,
120            mode_rx,
121        ));
122        let refresh_manager = Arc::new(RefreshManager::new(
123            Arc::clone(&pipeline_service),
124            metrics_service,
125            mode_tx,
126            commands,
127        ));
128        let pipeline_view_service = Arc::new(PipelineViewService::new(
129            view_repository,
130            Arc::clone(&provider_service),
131            Arc::clone(&pipeline_service),
132            view_config_state,
133            view_event_bus,
134        ));
135
136        Self {
137            provider_service,
138            pipeline_service,
139            refresh_manager,
140            health,
141            health_service,
142            log_service,
143            pipeline_view_service,
144        }
145    }
146}
pipedash_core/application/mod.rs

pipedash_core/application/
mod.rs
