Skip to main content

Observer

dataflow_rs::observe

Trait Observer 

Source 
pub trait Observer:
    Send
    + Sync
    + 'static {
    // Required method
    fn on(&self, event: &Event<'_>);

    // Provided method
    fn interest(&self) -> Interest { ... }
}
Expand description

What a host is told about a run as it goes: counters, histograms and spans, kept by the host.

An engine has one (EngineConfig::observer) and a run may add one beside it (RunOptions::observer). Each is called only with the events its interest names, and the clock is read only when one of them asks for Interest::TIMINGS.

on is called inline, synchronously, from the run: it must not block, re-enter the engine or panic. Push to a channel or bump an atomic and return.

use dataflow_rs::observe::{Event, Observer};
use dataflow_rs::run::EngineConfig;
use dataflow_rs::{Engine, Message};
use serde_json::json;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

#[derive(Default)]
struct Calls(AtomicU64);

impl Observer for Calls {
    fn on(&self, event: &Event<'_>) {
        if let Event::TaskFinished { .. } = event {
            self.0.fetch_add(1, Ordering::Relaxed);
        }
    }
}

let calls = Arc::new(Calls::default());
let engine = Engine::builder()
    .config(EngineConfig::default().observer(calls.clone()))
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.a", "logic": 1}]}}}
    ]}))
    .build()
    .unwrap();
engine.run(&mut Message::from_json(&json!({}))).await;
assert_eq!(calls.0.load(Ordering::Relaxed), 1);

Required Methods§

Source

fn on(&self, event: &Event<'_>)

One event of a run, inline and synchronously. It must not block or re-enter the engine; what it keeps, it copies.

Provided Methods§

Source

fn interest(&self) -> Interest

The events this observer is called with. Read once per run, when the run starts. WORKFLOWS | TASKS by default, which reads no clock.

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§