Struct Engine

Source

pub struct Engine { /* private fields */ }

Expand description

High-performance async workflow engine for message processing.

§Architecture

The engine is designed for async-first operation with Tokio:

Separation of Concerns: Distinct executors for workflows and tasks
Shared datalogic engine: Single datalogic_rs::Engine wrapped in Arc for thread-safe sharing
Arc: Pre-compiled logic shared across all async tasks
Async Functions: Native async support for I/O-bound operations

§Performance Characteristics

Zero Runtime Compilation: All logic compiled during initialization
Zero-Copy Sharing: Arc-wrapped compiled logic shared without cloning
Optimal for Mixed Workloads: Async I/O with blocking CPU evaluation
Thread-Safe by Design: All components safe to share across Tokio tasks

Implementations§

Source §

impl Engine

Source

pub fn new( workflows: Vec<Workflow>, task_functions: HashMap<String, BoxedFunctionHandler>, ) -> Result<Self>

Creates a new Engine instance.

Compiles every workflow / task / function-config JSONLogic expression up-front. Returns Err(DataflowError) if any required expression fails to compile — fail-loud at construction time instead of silently dropping broken workflows at runtime.

§Arguments

workflows - The workflows to use for processing messages
task_functions - Custom async function handlers (use HashMap::new() for none, or prefer Engine::builder)

§Example

use dataflow_rs::{Engine, Workflow};

let workflows = vec![Workflow::from_json(r#"{"id": "test", "name": "Test", "priority": 0, "tasks": [{"id": "task1", "name": "Task 1", "function": {"name": "map", "input": {"mappings": []}}}]}"#).unwrap()];

let engine = Engine::builder().with_workflows(workflows).build().unwrap();

The recommended construction path is Engine::builder. Engine::new is the lower-level escape hatch — accepts handlers as a plain HashMap (use HashMap::new() for the no-handler case).

Source

pub fn builder() -> EngineBuilder

Start building an engine. The recommended construction path — chains register("name", handler) and with_workflow(w) calls, then build() to produce a Result<Engine>.

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

Source

pub fn engine_version_value(&self) -> &OwnedDataValue

Cached OwnedDataValue::String of the engine version.

Source

pub fn with_new_workflows(&self, workflows: Vec<Workflow>) -> Result<Self>

Creates a new Engine with different workflows but the same custom function handlers.

This is the hot-reload path. The existing engine remains valid for any in-flight process_message calls. The returned engine shares the same function registry (zero-copy Arc bump) but has freshly compiled logic for the new workflow set.

§Arguments

workflows - The new set of workflows to compile and use

Source

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

Attach a per-task ExecutionObserver, returning the updated engine.

The escape hatch matching Engine::new — EngineBuilder::with_observer is the recommended path. Rebuilds the executor stack around the existing handler registry and datalogic engine, so nothing is recompiled; the cost is a few Arc bumps.

Carried across Engine::with_new_workflows, so a hot reload does not silently stop reporting.

Source

pub async fn process_message(&self, message: &mut Message) -> Result<()>

Processes a message through workflows that match their conditions.

This async method:

Iterates through workflows sequentially in priority order (pre-sorted at construction)
Delegates workflow execution to the WorkflowExecutor
Updates message metadata

§Error contract

Errors flow through two complementary channels:

message.errors() — always contains every error encountered (validation failures, task panics, 5xx-status outcomes, workflow wrappers). Callers that want a uniform view inspect this list.
Result::Err — signals only that the engine stopped before processing every workflow. Callers that want fail-fast match on this. The error pushed to message.errors for the same failure carries the workflow context (id) that the bare Err doesn’t.

In particular: a workflow with continue_on_error: true records its errors to message.errors and returns Ok(()) here. A workflow with continue_on_error: false records to message.errors and returns Result::Err (which short-circuits the rest of this call).

§Arguments

message - The message to process through workflows

§Returns

Result<()> — Ok(()) if every workflow completed (each may have pushed errors to message.errors); Err(e) if the engine stopped early on a hard failure.

Source

pub async fn process_message_tracing( &self, message: &mut Message, trace: &mut ExecutionTrace, ) -> Result<()>

Processes a message through workflows with step-by-step tracing, recording into a caller-owned trace.

Identical to Engine::process_message_with_trace except that the trace is borrowed rather than returned, so the steps completed before a hard failure survive the Err. That makes this the method to reach for when the run you want to inspect is the run that failed — a returned trace is dropped by the ? at the call site, a borrowed one is not.

Steps are appended to trace; any steps already present are preserved, so a caller can accumulate across a chain of calls.

The error contract is unchanged: Ok(()) means every workflow was processed (each may still have pushed to message.errors), and Err(e) means the engine stopped early. See Engine::process_message for the full contract.

Note that the failing task’s own step is not recorded — the engine propagates the failure before appending it — so the retained trace ends at the last known-good step rather than at the error. The error itself is available from the returned Err and from message.errors().

§Arguments

message - The message to process through workflows
trace - Caller-owned trace to append steps to

§Returns

Result<()> — Ok(()) if every workflow completed; Err(e) if the engine stopped early. In both cases trace holds the steps that ran.

Source

pub async fn process_message_with_trace( &self, message: &mut Message, ) -> Result<ExecutionTrace>

Processes a message through workflows with step-by-step tracing.

This method is similar to process_message but captures an execution trace that can be used for debugging and step-by-step visualization.

Because the trace is returned by value, a ? at the call site discards it — on a hard failure this yields Err and no steps at all. Use Engine::process_message_tracing to keep the steps that ran.

§Arguments

message - The message to process through workflows

§Returns

Result<ExecutionTrace> - The execution trace with message snapshots

Source

pub async fn process_message_with_trace_options( &self, message: &mut Message, options: TraceOptions, ) -> Result<ExecutionTrace>

Processes a message with tracing under an explicit capture policy.

The default policy — what Engine::process_message_with_trace uses — takes a full Message snapshot per executed step, which is unbounded in message size and quadratic in task count. A host that persists traces should bound them here rather than trimming the result afterwards; by then the peak memory has already been paid.

See TraceOptions for the knobs, and Engine::process_message_tracing if you also need the steps to survive a hard failure.

§Arguments

message - The message to process through workflows
options - What to record for each step

Source

pub async fn process_message_for_channel( &self, channel: &str, message: &mut Message, ) -> Result<()>

Processes a message through only the Active workflows registered for a given channel.

Workflows are processed in priority order (lowest first), same as process_message(). If the channel does not exist or has no Active workflows, this is a no-op.

§Arguments

channel - The channel name to route the message through
message - The message to process

Source

pub async fn process_message_for_channel_tracing( &self, channel: &str, message: &mut Message, trace: &mut ExecutionTrace, ) -> Result<()>

Channel-scoped variant of Engine::process_message_tracing.

As with Engine::process_message_for_channel, an unknown channel — or a channel with no Active workflows — is a no-op: this returns Ok(()) and leaves trace untouched. Steps are appended, matching Engine::process_message_tracing.

§Arguments

channel - The channel name to route the message through
message - The message to process
trace - Caller-owned trace to append steps to

Source

pub async fn process_message_for_channel_with_trace( &self, channel: &str, message: &mut Message, ) -> Result<ExecutionTrace>

Processes a message through a channel with step-by-step tracing.

Because the trace is returned by value, a ? at the call site discards it — on a hard failure this yields Err and no steps at all. Use Engine::process_message_for_channel_tracing to keep the steps that ran.