datum-agent 0.10.0

Embeddable Datum job registry and lifecycle supervisor
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

//! Compile-checked snippet backing docs/guides/datum-agent.md.
//! The region between the markers is imported verbatim via the VitePress `<<<` include.

use std::time::Duration;

#[test]
fn agent_quickstart_example() {
    // #region agent-quickstart
    use datum::prelude::*;
    use datum_agent::{Agent, JobMat, JobSpec};

    // Start an embedded agent and get its job registry.
    let agent = Agent::start().expect("agent starts");
    let registry = agent.registry();

    // A job is a NAMED, SUPERVISED stream: the factory rebuilds the blueprint on
    // every (re)start, and the agent wires in drain + instrumentation for you.
    let spec = JobSpec::new("tick-counter", |context| {
        let control = context.control();
        Ok(
            Source::tick(Duration::ZERO, Duration::from_millis(10), 1_u64)
                .instrumented(
                    format!("{}:{}", context.name(), context.generation()),
                    context.instrumentation_registry(),
                )
                .via_mat(context.drain_flow(), Keep::right)
                .to_mat(Sink::ignore(), move |_switch, completion| {
                    JobMat::new(completion, control.clone())
                }),
        )
    });

    registry.submit(spec).expect("job submitted");
    registry.start("tick-counter").expect("job started");

    let status = registry.status("tick-counter").expect("status");
    assert_eq!(status.name, "tick-counter");

    // Graceful drain: stop intake, let in-flight elements finish, then stop.
    registry.drain("tick-counter").expect("drain requested");
    registry.shutdown().expect("agent shuts down");
    // #endregion agent-quickstart
}