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

pipedash_core/application/
mod.rs
