pipedash-core 0.7.0

Framework-agnostic core library for Pipedash
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

mod config_watcher;
pub mod fetch_scheduler;
pub mod services;
#[cfg(test)]
pub(crate) mod test_support;

use std::sync::Arc;

pub use config_watcher::ConfigWatcher;
pub use fetch_scheduler::{
    AttemptHandle,
    BackoffPolicy,
    ClaimKind,
    FetchClaim,
    FetchResult,
    FetchScheduler,
    ForcedClaim,
    Generation,
    Outcome,
    ScheduleView,
};
pub use services::health_service::{
    refresh_notifications,
    HealthHandle,
    HealthInput,
    HealthReceivers,
    HealthSnapshot,
    LocalNotification,
    NotificationState,
    ProviderHealthInput,
    ProviderHealthService,
    ProviderHealthState,
};
pub use services::log_service::{
    LogExport,
    LogService,
};
pub use services::metrics_service::MetricsService;
pub use services::pipeline_service::{
    PipelineService,
    RunWatch,
    WatchedRun,
};
pub use services::pipeline_view_service::PipelineViewService;
pub use services::provider_service::{
    ProviderLease,
    ProviderService,
};
use tokio::sync::watch;

use crate::event::EventBus;
use crate::infrastructure::database::Repository;
use crate::infrastructure::run_logs::LogStore;
use crate::infrastructure::{
    ConfigState,
    HttpClientManager,
};

mod refresh_manager;
pub use refresh_manager::{
    RefreshCommand,
    RefreshDispatcher,
    RefreshManager,
    RefreshMode,
};

pub struct Services {
    pub provider_service: Arc<ProviderService>,
    pub pipeline_service: Arc<PipelineService>,
    pub refresh_manager: Arc<RefreshManager>,
    pub health: HealthHandle,
    pub health_service: Arc<ProviderHealthService>,
    pub log_service: Arc<LogService>,
    pub pipeline_view_service: Arc<PipelineViewService>,
}

impl Services {
    pub fn new(
        repository: Arc<Repository>, http_client_manager: Arc<HttpClientManager>,
        metrics_service: Option<Arc<MetricsService>>, event_bus: Arc<dyn EventBus>,
        config_state: Arc<ConfigState>, log_store: Arc<LogStore>,
    ) -> Self {
        let (health, health_receivers) = HealthHandle::channel();
        let (dispatcher, commands) = RefreshDispatcher::channel();
        let (mode_tx, mode_rx) = watch::channel(RefreshMode::Active);
        config_state.attach_health(health.clone());
        let view_repository = Arc::clone(&repository);
        let view_config_state = Arc::clone(&config_state);
        let view_event_bus = Arc::clone(&event_bus);

        let provider_service = Arc::new(ProviderService::new(
            Arc::clone(&repository),
            Arc::clone(&http_client_manager),
            Arc::clone(&event_bus),
            Arc::new(FetchScheduler::default()),
            health.clone(),
            dispatcher,
            Arc::clone(&config_state),
            Arc::clone(&log_store),
        ));
        let health_service = Arc::new(ProviderHealthService::new(
            health_receivers,
            Arc::clone(&provider_service),
            http_client_manager,
            Arc::clone(&event_bus),
            config_state,
            mode_tx.subscribe(),
        ));
        let log_service = Arc::new(LogService::new(
            log_store,
            Arc::clone(&provider_service),
            Arc::clone(&repository),
            Arc::clone(&event_bus),
        ));
        let pipeline_service = Arc::new(PipelineService::new(
            repository,
            Arc::clone(&provider_service),
            metrics_service.clone(),
            event_bus,
            mode_rx,
        ));
        let refresh_manager = Arc::new(RefreshManager::new(
            Arc::clone(&pipeline_service),
            metrics_service,
            mode_tx,
            commands,
        ));
        let pipeline_view_service = Arc::new(PipelineViewService::new(
            view_repository,
            Arc::clone(&provider_service),
            Arc::clone(&pipeline_service),
            view_config_state,
            view_event_bus,
        ));

        Self {
            provider_service,
            pipeline_service,
            refresh_manager,
            health,
            health_service,
            log_service,
            pipeline_view_service,
        }
    }
}