pipedash-core 0.4.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

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

use std::sync::Arc;

pub use fetch_scheduler::{
    AttemptHandle,
    BackoffPolicy,
    ClaimKind,
    FetchClaim,
    FetchResult,
    FetchScheduler,
    ForcedClaim,
    Generation,
    Outcome,
    ScheduleView,
};
pub use services::health_service::{
    HealthHandle,
    HealthInput,
    HealthReceivers,
    HealthSnapshot,
    ProviderHealthInput,
    ProviderHealthService,
    ProviderHealthState,
};
pub use services::metrics_service::MetricsService;
pub use services::pipeline_service::PipelineService;
pub use services::provider_service::{
    ProviderLease,
    ProviderService,
};
use tokio::sync::watch;

use crate::event::EventBus;
use crate::infrastructure::database::Repository;
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>,
}

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>,
    ) -> 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 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,
        ));
        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 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,
        ));

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