Struct EngineBuilder 

pub struct EngineBuilder { /* private fields */ }

Builder for Engine. The recommended construction path — chain register("name", handler) and with_workflow(workflow) calls, then build() to produce a Result<Engine>. Empty registration is fine; an engine with no custom handlers still resolves the built-in functions.

register takes any AsyncFunctionHandler and boxes it internally; the Box<dyn DynAsyncFunctionHandler + Send + Sync> plumbing stays out of user code.

use dataflow_rs::{Engine, Workflow};
let engine = Engine::builder()
    .with_workflow(workflow)
    // .register("my_handler", MyHandler)
    .build()
    .unwrap();

Implementations

impl EngineBuilder

pub fn new() -> Self

Create an empty builder. Equivalent to EngineBuilder::default.

pub fn register<F>(self, name: impl Into<String>, handler: F) -> Self

where F: AsyncFunctionHandler,

Register a custom async handler under name. Accepts any AsyncFunctionHandler; boxing happens internally via the engine’s blanket impl.

pub fn register_boxed( self, name: impl Into<String>, handler: BoxedFunctionHandler, ) -> Self

Register a pre-boxed handler. Useful when handlers are constructed dynamically (e.g. plugin registries) and the concrete type isn’t known at the call site.

pub fn dispatchable_functions( &self, ) -> impl Iterator<Item = DispatchableFunction<'_>>

Every function this builder will dispatch once built.

The pre-build twin of Engine::dispatchable_functions, with identical semantics — the two agree by construction, since build() moves this registry into the engine unchanged. Takes &self, so screening a batch of definitions does not consume the builder.

use dataflow_rs::Engine;

let builder = Engine::builder();
let names: Vec<&str> = builder.dispatchable_functions().map(|f| f.name).collect();

assert!(names.contains(&"parse_json"));
assert!(!names.contains(&"publish_kafka")); // config schema, no handler

pub fn can_dispatch(&self, name: &str) -> bool

Whether the engine this builder produces will run a task named name.

The pre-build twin of Engine::can_dispatch. Screening a workflow is then a filter over its tasks — note that Workflow::tasks is already flattened, so this covers members of task groups too:

use dataflow_rs::{Engine, Workflow};

let workflow = Workflow::from_json(r#"{
    "id": "w", "name": "w", "priority": 0,
    "tasks": [
        {"id": "a", "name": "a", "function": {"name": "map", "input": {"mappings": []}}},
        {"id": "b", "name": "b",
         "function": {"name": "enrich",
                      "input": {"connector": "c", "merge_path": "data.out"}}}
    ]
}"#).unwrap();

let builder = Engine::builder();
let unrunnable: Vec<&str> = workflow
    .tasks
    .iter()
    .map(|t| t.function.function_name())
    .filter(|name| !builder.can_dispatch(name))
    .collect();

assert_eq!(unrunnable, vec!["enrich"]);

pub fn check_workflow(&self, workflow: &Workflow) -> Vec<WorkflowIssue>

Check a workflow against this builder’s registered handlers and operators, without consuming the builder or building an engine.

The pre-build twin of Engine::check_workflow, with identical semantics. Takes &self, so a host can screen a batch of definitions against the registrations it is about to build with.

Templates are compiled against a datalogic engine configured exactly as Self::build will configure it — same custom operators, same templating mode — so a template that passes here compiles there.

use dataflow_rs::{Engine, IssueCode, Workflow};

let workflow = Workflow::from_json(r#"{
    "id": "w", "name": "w", "priority": 0,
    "tasks": [{"id": "t", "name": "t",
               "function": {"name": "typo_handler", "input": {}}}]
}"#).unwrap();

let issues = Engine::builder().check_workflow(&workflow);
assert_eq!(issues[0].code, IssueCode::UnknownFunction);
assert_eq!(issues[0].task_id.as_deref(), Some("t"));

pub fn with_workflow(self, workflow: Workflow) -> Self

Add a single workflow. Subsequent calls append.

pub fn with_workflows<I>(self, workflows: I) -> Self

where I: IntoIterator<Item = Workflow>,

Append every workflow in workflows. Accepts anything iterable — Vec<Workflow>, an array, an iterator. Existing workflows on the builder are kept; subsequent registers/workflows still chain.

pub fn with_handlers( self, handlers: HashMap<String, BoxedFunctionHandler>, ) -> Self

Insert every handler in handlers, keeping any already registered.

Same extend-not-replace semantics as EngineBuilder::with_workflows. Exists because register is per-name, which pushed an embedder that builds a whole HashMap<String, BoxedFunctionHandler> in one place onto Engine::new and off the builder entirely — and therefore out of reach of EngineBuilder::with_observer.

pub fn with_observer(self, observer: Arc<dyn ExecutionObserver>) -> Self

Attach a per-task ExecutionObserver. Later calls replace the previous one.

This is the only way to time the sync built-ins, which are dispatched inside the executor and never reach the function registry. With no observer attached the instrumentation — including its clock reads — stays out of the dispatch path entirely.

pub fn with_error_context_path(self, path: impl Into<String>) -> Self

Mirror per-task failure codes into the message context at path, so a downstream condition or map can branch on why a task failed.

Off unless called: with no path configured nothing is written and the mechanism costs one Option check on a path that only runs after a task has already failed.

One record is appended per error a task contributes to Message::errors:

{ "workflow_id": "place_order", "task_id": "charge_payment",
  "code": "TIMEOUT_ERROR", "status": 500 }

so a later task can gate on the reason:

{ "in": [ { "var": "metadata.errors.0.code" },
          ["TIMEOUT_ERROR", "IO_ERROR"] ] }

Coverage matches errors() exactly — a handler returning Err, a task returning a 5xx outcome, the validation built-in’s per-rule failures, and anything a handler adds through TaskContext::add_error all appear. The workflow-level WORKFLOW_ERROR wrapper does not: it re-reports the same underlying failure, so mirroring it would double-count.

status is the task’s own status — 500 when the handler returned Err, otherwise the status the outcome carried (400 for validation). That is the distinction metadata.progress cannot make, since its failure arm hard-codes 500.

The error message and the operator-only detail are deliberately not recorded: the context is serialized back to callers, and detail is documented as unsafe to hand to an untrusted one. Read those from message.errors() host-side.

path must start with data, metadata or temp_data — the JSONLogic evaluation context is exactly those three slots — and may not be metadata.progress. Violations fail EngineBuilder::build.

pub fn with_error_context_limit(self, limit: usize) -> Self

Cap the number of records retained at the error-context path, keeping the most recent (default 32).

The bound is what keeps the option’s memory cost independent of a looping workflow’s iteration count: Message.context is deep-cloned into every trace snapshot, so an uncapped list in a loop with a failing body grows the trace quadratically. Conditions overwhelmingly read the latest failure, so the oldest records are the ones dropped.

Setting a limit without a path is inert, not an error. A limit of 0 fails EngineBuilder::build.

pub fn with_datalogic_operator<T>( self, name: impl Into<String>, operator: T, ) -> Self

where T: CustomOperator + 'static,

Register a custom JSONLogic operator on the engine’s internal datalogic instance, under name. Later calls with the same name replace the earlier registration.

This is the host’s door for domain operators: the engine builds (and on Engine::with_new_workflows rebuilds) its datalogic engine internally, where registration is builder-only — so operators must enter here to exist at all, and are retained on the engine so every hot reload re-registers them.

Semantics follow datalogic_rs: arguments arrive pre-evaluated, and a built-in operator name always wins over a custom registration — pick names no built-in uses. Because the engine always runs in templating mode, a name that is not registered is not an error: the object echoes back as literal data, exactly like a disabled operator family. Registering a name therefore converts previously-inert values into live operator calls, the same caveat the cargo features carry.

pub fn build(self) -> Result<Engine>

Compile the workflows, pre-parse Custom inputs, and produce the engine. Compile errors and missing handler references surface here — the engine never deserializes Custom config on the hot path.

Trait Implementations

impl Default for EngineBuilder

fn default() -> EngineBuilder

Returns the “default value” for a type.