Struct Engine

Source

pub struct Engine { /* private fields */ }

Expand description

A program and its config: what runs messages.

An Arc inside, so cloning is cheap and a clone runs the same program. with_program is a reload: the engine it returns runs the new program, and this one, with any run in flight, keeps the old.

use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::{Engine, Message};
use serde_json::json;

let engine = Engine::builder()
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.greeting", "logic": "hello"}]}}}
    ]}))
    .build()
    .unwrap();
let mut message = Message::from_json(&json!({}));
let run = engine.run(&mut message).await;
assert_eq!(run.matched(), 1);
let greeting = OwnedDataValue::from(&json!({"greeting": "hello"}));
assert_eq!(message.context().data(), &greeting);

Implementations§

Source §

impl Engine

Source

pub fn new(program: Program, cfg: EngineConfig) -> Self

An engine running program with cfg.

Source

pub fn builder() -> EngineBuilder

A builder starting from the default registry, for a host that does not quarantine: its build fails if any workflow is rejected.

Source

pub async fn run(&self, msg: &mut Message) -> Run

Runs msg with RunOptions::default.

Source

pub async fn run_with(&self, msg: &mut Message, opts: &RunOptions<'_>) -> Run

Runs msg through the program’s Active workflows, those of opts.channel when it names one, in priority order.

The run reads the clock once: metadata.processed_at, every audit entry and every error record carry that reading. Only an observer asking for TIMINGS, or a trace asking for timings, has the clock read again, for the elapsed times it is given. It writes metadata.engine_version, and metadata.channel under a channel, before any workflow runs, and then, with an ErrorContext configured, removes its path’s key: the records there are the run’s own. A workflow whose rollout does not admit opts.routing_bucket is skipped; no bucket is admitted by every rollout. A workflow that stops the run ends it, and the Run says how: TimedOut once opts.deadline passed, Cancelled once opts.cancel was cancelled, and otherwise Stopped with the error that stopped it, whatever its kind: a child run’s timeout that a function returns before this run’s own deadline passed is its task’s failure, not this run’s. An Err a function returns once opts.deadline has passed, or opts.cancel was cancelled, is the run’s limit’s.

With opts.trace set, the Run carries the run’s Trace: the context after the stamps and the reset, then a step for every recorded call, skipped call or workflow and sweep, the step that stopped the run included: the call a limit cut short, or, for a stop as a sweep starts, the engine’s step for the counter it wrote. Its replay is the message’s context on every status.

Source