datum_agent/
agent.rs

1use std::time::Duration;
2
3use datum::StreamInstrumentationRegistry;
4
5use crate::{AgentResult, JobRegistry, JobRegistryHandle};
6
7/// Configuration for an embeddable Datum agent.
8#[derive(Debug, Clone)]
9pub struct AgentConfig {
10    /// How often the registry polls `StreamCompletion::try_wait()` for running jobs.
11    pub poll_interval: Duration,
12    /// Per-subscriber lifecycle event buffer capacity.
13    pub event_buffer: usize,
14    /// Registry sampled by DCP metrics subscriptions.
15    pub instrumentation: StreamInstrumentationRegistry,
16}
17
18impl Default for AgentConfig {
19    fn default() -> Self {
20        Self {
21            poll_interval: Duration::from_millis(10),
22            event_buffer: 1_024,
23            instrumentation: StreamInstrumentationRegistry::new(),
24        }
25    }
26}
27
28/// Minimal embeddable entrypoint for the local agent.
29pub struct Agent;
30
31impl Agent {
32    pub fn start() -> AgentResult<AgentHandle> {
33        Self::start_with_config(AgentConfig::default())
34    }
35
36    pub fn start_with_config(config: AgentConfig) -> AgentResult<AgentHandle> {
37        let instrumentation = config.instrumentation.clone();
38        Ok(AgentHandle {
39            registry: JobRegistry::start(config)?,
40            instrumentation,
41        })
42    }
43}
44
45/// Handle returned by [`Agent::start`] and [`Agent::start_with_config`].
46#[derive(Clone)]
47pub struct AgentHandle {
48    registry: JobRegistryHandle,
49    instrumentation: StreamInstrumentationRegistry,
50}
51
52impl AgentHandle {
53    #[must_use]
54    pub fn registry(&self) -> &JobRegistryHandle {
55        &self.registry
56    }
57
58    #[must_use]
59    pub fn instrumentation_registry(&self) -> &StreamInstrumentationRegistry {
60        &self.instrumentation
61    }
62
63    pub fn into_registry(self) -> JobRegistryHandle {
64        self.registry
65    }
66}
datum_agent/agent.rs

datum_agent/
agent.rs
