API Reference

Dataflow-rs 4.0 is organized into seven public modules with sixteen core types re-exported at the crate root. This page provides an overview of each module, the purpose of every public type, and common usage examples. Full API documentation is available on docs.rs.

At a high level:

1. Workflow JSON is parsed and linted by def::parse.
2. Parsed definitions are compiled against a Registry by program::Compiler.
3. Messages are executed by run::Engine over a run::Message.
4. Execution is observed via observe::Observer or observe::Trace.
5. Custom functions implement function::Function.

Crate Root Re-exports

The crate root re-exports the sixteen most commonly used types and traits:

In addition, VERSION provides the crate version string (written to metadata.engine_version on every run), and datalogic_rs and datavalue are re-exported so downstream crates don't need to manage separate dependency versions.

def — Workflow Definitions

Contains the types representing authored workflow definitions. Definitions are parsed and linted via def::parse (or def::parse_owned) and serialize back to canonical 4.0 JSON. (docs.rs)

expr — Expressions, Paths, and Secrets

Provides the JSONLogic compilation and evaluation engine, context path handling, secret store, and configuration field types. (docs.rs)

function — Custom Host Functions and Layers

Defines the Function trait, configuration derive macro, call context, middleware layers, and function testing utilities. (docs.rs)

program — Compilation and Validation

Compiles parsed WorkflowDefs against a Registry into an immutable Program. (docs.rs)

run — Executing Messages

Executes compiled programs over messages. (docs.rs)

observe — Observers and Traces

Provides inline event observation and step-by-step execution traces. (docs.rs)

compat — 3.x Migration Utilities

Provides tools for upgrading 3.x workflow definitions to canonical 4.0 JSON. (docs.rs)

See Migrating from 3.x for details.

Common Usage Patterns

Building an Engine

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;
use dataflow_rs::run::EngineBuilder;
use serde_json::json;

let builder: EngineBuilder = Engine::builder();
let engine: Engine = builder
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.greeting", "logic": "hello"}]}}}
    ]}))
    .build()
    .expect("Err contains all issues from rejected definitions");
assert_eq!(engine.program().workflows().len(), 1);
}

EngineBuilder supports .register(name, function), .layer(l), .registry(r), .options(CompileOptions), .config(EngineConfig), .workflow(WorkflowDef), .workflow_json(&Value), and .workflow_json_owned(Value), followed by .build() -> Result<Engine, Vec<Issue>>.

If your application needs to quarantine invalid workflows and run the valid ones, compile workflows individually with Compiler:

#![allow(unused)]
fn main() {
use dataflow_rs::program::{CompileOptions, Compiler, Program};
use dataflow_rs::run::EngineConfig;
use dataflow_rs::{Engine, Registry};
use dataflow_rs::def::{ParseOptions, parse};
use serde_json::json;
let def = parse(&json!({"id": "w", "tasks": [
    {"id": "t", "function": {"name": "log", "input": {"message": "hi"}}}
]}), ParseOptions::default()).def.unwrap();

let compiler: Compiler = Registry::default().compiler(CompileOptions::default());
let compiled = compiler.compile_one(&def).expect("Err contains all issues");
let program: Program = compiler.link([compiled.workflow]);
let engine = Engine::new(program, EngineConfig::default());
assert_eq!(engine.program().workflows()[0].id, "w");
}

To reload workflows on a running engine without disrupting in-flight runs, call engine.compiler(), link the new workflows, and call engine.with_program(program).

Running a Message

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::{Engine, Message, Run, RunOptions};
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!({"name": "Ada"}));

// Run with default options:
let run: Run = engine.run(&mut message).await;

// Run with custom per-run options (channel, routing bucket, deadline, trace, etc.):
let opts = RunOptions::default().channel("default");
let run: Run = engine.run_with(&mut message, &opts).await;
assert_eq!(run.matched(), 1);
});
}

Inspecting the Run Report

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::run::RunStatus;
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;

match run.status() {
    RunStatus::Completed => {}
    RunStatus::CompletedWithErrors => { /* inspect message.errors() */ }
    RunStatus::Stopped(error) => eprintln!("{}: {}", error.code(), error.message()),
    RunStatus::TimedOut | RunStatus::Cancelled => {}
    _ => {}
}
for workflow in run.workflows() {
    println!("{} {:?} {:?}", workflow.id, workflow.selection, workflow.exit);
}
assert!(matches!(run.status(), RunStatus::Completed));
});
}

If you prefer a Result<(), Error> that returns Err when a run stops, times out, or is cancelled, call run.into_result() (extracting run.take_trace() first if you need the trace).

Constructing a Message

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use dataflow_rs::datavalue::OwnedDataValue;
use serde_json::json;
use std::sync::Arc;

// From a serde_json::Value payload:
let m = Message::from_json(&json!({"name": "Ada"}));

// From an OwnedDataValue or a shared Arc<OwnedDataValue>:
let payload = Arc::new(OwnedDataValue::from(&json!({"name": "Ada"})));
let m = Message::new(Arc::clone(&payload));

// With pre-populated `data` and `metadata` objects:
let m = Message::from_parts(
    payload,
    OwnedDataValue::from(&json!({"stage": "intake"})),
    OwnedDataValue::Null,
);
assert_eq!(m.context().data(), &OwnedDataValue::from(&json!({"stage": "intake"})));
}

Implementing a Custom Host Function

#![allow(unused)]
fn main() {
use dataflow_rs::expr::Output;
use dataflow_rs::function::FunctionConfig;
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::{Ctx, Expr, Function, Outcome, Result};
use serde_json::json;

#[derive(FunctionConfig)]
struct FetchConfig {
    url: Expr<String>,
    #[output(default = "data.response")]
    output: Output,
}

struct Fetch;

impl Function for Fetch {
    type Config = FetchConfig;

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &FetchConfig) -> Result<Outcome> {
        let url = ctx.eval(&cfg.url)?;
        let response = OwnedDataValue::from(&json!({"url": &*url, "status": 200}));
        ctx.emit(&cfg.output, response)?;
        Ok(Outcome::Ok)
    }
}
let _ = Fetch;
}

Validating a Workflow Definition

#![allow(unused)]
fn main() {
use dataflow_rs::def::{ParseOptions, Parsed, parse};
use dataflow_rs::program::{CompileOptions, Severity};
use dataflow_rs::{Issue, Registry};
use serde_json::json;

let json = json!({"id": "w", "tasks": [
    {"id": "t", "function": {"name": "map", "input": {
        "mappings": [{"path": "data.x", "logic": {"var": "payload.x"}}]}}}
]});

// 1. Validate structure (`def` is `Some` when no issue is `Rejected`):
let parsed: Parsed = parse(&json, ParseOptions::default());
let def = parsed.def.expect("the structure is fine");

// 2. Validate functions, inputs, expressions, paths, and secrets:
let compiler = Registry::default().compiler(CompileOptions::default());
let issues: Vec<Issue> = compiler.check(&def);
let rejected = issues.iter().filter(|i| i.severity == Severity::Rejected).count();
assert_eq!(rejected, 0);
assert_eq!(issues[0].code.as_str(), "PAYLOAD_VAR");
assert_eq!(issues[0].path.to_string(), "tasks[0].function.input.mappings[0].logic");
}

Error Codes

Error::code() returns the error code string written to ErrorRecords on message.errors() and reported on RunStatus::Stopped(error):

In addition, the engine records the following codes directly on message.errors() without producing a standalone Error:

Cargo Features

No Cargo features are enabled by default.

[dependencies]
dataflow-rs = { version = "4.0", features = ["datetime", "budget"] }