Integration Functions

The http_call, enrich, and publish_kafka functions provide typed configuration schemas for the three most common service-layer integration patterns. Unlike map or validation, they do not ship with a built-in handler — the actual I/O is provided by your application via AsyncFunctionHandler.

Why a config schema without an implementation?

The engine itself is I/O-agnostic: it doesn’t bundle an HTTP client, a Kafka producer, or any other transport. But the shape of these integrations is predictable enough that dataflow-rs provides typed config structs so that:

• JSONLogic expressions inside the config — since 3.9, that is every parameter are pre-compiled at engine startup — same fail-loud behaviour as map rules
• Misshapen config fails at Engine::new(), not at first message
• Your handler receives an already-validated HttpCallConfig / EnrichConfig / PublishKafkaConfig — no per-call JSON parse

How to use them

For each integration variant you want to use, register a handler under the matching name when building the engine:

use dataflow_rs::prelude::*;
use dataflow_rs::HttpCallConfig;
use async_trait::async_trait;

struct HttpCallHandler { /* reqwest::Client, connector registry, etc. */ }

#[async_trait]
impl AsyncFunctionHandler for HttpCallHandler {
    type Input = HttpCallConfig;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        cfg: &HttpCallConfig,
    ) -> Result<TaskOutcome> {
        // Every parameter is JSONLogic, so read each through its `resolve_*`
        // method. A statically-authored one is already computed; a dynamic one
        // evaluates on the worker thread's pooled arena.
        let connector = cfg.resolve_connector(ctx)?;
        let path = cfg.resolve_path(ctx)?;
        let headers = cfg.resolve_headers(ctx)?;
        let body = cfg.resolve_body(ctx)?;
        let timeout_ms = cfg.resolve_timeout_ms(ctx)?;
        let method = cfg.method.as_str();       // canonical token for your client

        // Resolve `connector` against your own registry, make the call, then
        // merge the response into ctx at cfg.resolve_response_path(ctx)?…
        let _ = (connector, path, headers, body, timeout_ms, method);
        Ok(TaskOutcome::Success)
    }
}

let engine = Engine::builder()
    .register("http_call", HttpCallHandler { /* … */ })
    .with_workflow(workflow)
    .build()?;

Skip the registration step and any workflow that uses these variants will fail with DataflowError::FunctionNotFound("http_call") at dispatch time.

Every parameter is JSONLogic

Since 3.9 every field of every integration config is an expression, and the engine compiles it at build(). Read them through the resolve_* methods:

The static spelling is unchanged and costs nothing. A JSON literal is JSONLogic for itself, so "connector": "user_service" and "timeout_ms": 5000 mean exactly what they did. Those fold to a constant at build() and are cached, so only a parameter that actually reads the message does per-message work.

What the methods guarantee that a hand-rolled read does not:

• An evaluation failure propagates as DataflowError::LogicEvaluation rather than substituting something else — a different URL because an expression errored would hide a real problem.
• Path, key, header and connector results are coerced to a plain string — a number becomes its digits, a container its compact JSON — because those values go into a URL, a header or a partition key. resolve_value deliberately returns Option<Value> instead, so a producer that serializes unconditionally is not forced through the key’s coercion and end up with different bytes on the wire.
• A failing header value fails the whole call rather than sending the request with that header missing, which would surface as a confusing 401.

Migrating from *_logic

path/path_logic and body/body_logic were separate fields because a single field could not be either a literal or an expression: a literal object whose key matched an operator name would evaluate. The $ escape removes that, so the pairs collapsed:

The old names are kept as serde aliases, so existing workflow definitions load unchanged. Supplying both spellings is a duplicate field error rather than a precedence rule.

One case needs the escape: a literal object body with a field named after an operator. Write {"$cat": …} for a body field actually called cat.

Each *_logic field is a Template — the same type available for your own handler’s config. There is no separate compiled slot to read directly; resolve_* is the only supported way to get a value out of one.

Detecting a missing handler before it fails

Because these three names deserialize into typed built-in variants, a workflow that uses one without a registered handler builds cleanly — Engine::new() raises nothing, and the failure arrives on the first message. That is deliberate: a host screening stored workflow definitions one row at a time should not be stopped from booting by a single unusable row.

To detect the gap instead of discovering it at runtime, classify the name:

#![allow(unused)]
fn main() {
use dataflow_rs::{BuiltinKind, builtin_function_kind};

// Executed by the crate — always runnable, no registration needed.
assert_eq!(builtin_function_kind("map"), Some(BuiltinKind::SelfContained));

// Config schema only — needs a handler registered under the same name.
assert_eq!(
    builtin_function_kind("enrich"),
    Some(BuiltinKind::RequiresHandler),
);

// Not a built-in at all — lands in `FunctionConfig::Custom`.
assert_eq!(builtin_function_kind("my_handler"), None);
}

That tells you a name needs a handler. It cannot tell you whether one is registered — for that, ask the engine or the builder directly.

Asking whether a name will actually run

can_dispatch answers the whole question in one call:

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

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

// Executed by the crate itself.
assert!(engine.can_dispatch("map"));
assert!(engine.can_dispatch("validation")); // alias of `validate`

// Config schema with nothing behind it — this is the case that builds
// cleanly and then fails every message.
assert!(!engine.can_dispatch("enrich"));
}

The guarantee runs both ways: a name can_dispatch accepts will execute, and a name it rejects fails with FunctionNotFound on the first message that reaches it. So screening a definition is a filter over its tasks — and because Workflow::tasks is already flattened, this covers tasks inside groups too:

#![allow(unused)]
fn main() {
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"]);
}

Both Engine and EngineBuilder carry the method, with identical semantics — check before you build, or against the engine you are already running.

Enumerating the whole vocabulary

dispatchable_functions() lists every name that will run, for completion tooling, an admin catalogue, or a did-you-mean on an unknown name:

#![allow(unused)]
fn main() {
use dataflow_rs::{BuiltinKind, Engine};

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

let validate = engine
    .dispatchable_functions()
    .find(|f| f.name == "validate")
    .unwrap();

assert_eq!(validate.kind, Some(BuiltinKind::SelfContained));
assert_eq!(validate.aliases, &["validation"]);
}

Three things to know about the result:

• Aliases are grouped. validate is yielded once carrying ["validation"], not twice. can_dispatch still accepts either spelling, so the two are deliberately different sets.
• kind is Option<BuiltinKind>. None means a registered custom handler, matching what builtin_function_kind already returns for a non-built-in name.
• Ordering is not meaningful. Treat it as a set; collect and sort if you need stable output.

Registering a handler under a SelfContained name is inert — map deserializes to the crate’s own implementation, which never consults the registry — so such a name still appears exactly once, as a built-in.

Prefer all of these over parsing the text of FunctionNotFound, which is a human-facing diagnostic and may be reworded at any time.

http_call

Issue an HTTP request and optionally merge the response into the message context.

Configuration

{
    "function": {
        "name": "http_call",
        "input": {
            "connector": "user_service",
            "method": "GET",
            "path": { "cat": ["/users/", {"var": "data.user_id"}] },
            "headers": {
                "X-Request-Id": { "var": "metadata.request_id" },
                "Authorization": { "cat": ["Bearer ", {"secret": "user_service_token"}] },
                "X-Env": "prod"
            },
            "response_path": "data.user_profile",
            "timeout_ms": 5000
        }
    }
}

Parameters

Every parameter except method is JSONLogic. A plain string or number is a literal, so the static spelling below is unchanged and costs nothing — see Every parameter is JSONLogic.

A header value is where a credential belongs: {"secret": "name"} reads the engine’s secret store, which no message ever carries. Before 3.9 header values were plain strings, so a token had to be injected by the service layer.

body_format and response_format are data, not API surface: dataflow-rs carries them but neither validates nor interprets their values — the service layer that implements http_call owns the value table, the default for an absent field, and the encoding behaviour. The split is deliberate: field names are fixed here by deny_unknown_fields, but a service layer can grow new values (say, "multipart") without a release of this crate.

response_path accepts output as an alias, so a service layer can present one destination-field name across its whole function catalogue. Supplying both keys is a duplicate field error rather than a precedence rule.

The alias is specific to http_call. enrich names its destination merge_path and publish_json / publish_xml name theirs target; neither takes output.

Unknown fields are rejected

All three integration configs reject keys they do not recognise. A misspelled field used to parse cleanly and be discarded, so an http_call task would make its request and silently throw the response away — no error at Engine::builder().build(), none at dispatch. Now it fails at parse time:

config for function 'http_call': unknown field `outputs`, expected one of
`connector`, `method`, `path`, `path_logic`, `headers`, `body`, `body_logic`,
`body_format`, `output`, `response_path`, `response_format`, `timeout_ms`

(`path_logic` and `body_logic` appear because they are still accepted as
back-compat aliases for `path` and `body`.)

Note this fails when the workflow definition is parsed, so a host loading stored definitions row by row sees one bad row fail its own parse rather than losing the whole set.

Converting method for your HTTP client

This crate takes no HTTP-client dependency, so your handler converts HttpMethod into whatever type its client uses. as_str() gives the canonical token — the same spelling the config accepts — so the bridge is one line and needs no match:

#![allow(unused)]
fn main() {
use dataflow_rs::HttpMethod;

// e.g. reqwest::Method::from_bytes(m.as_str().as_bytes())
assert_eq!(HttpMethod::Patch.as_str(), "PATCH");
assert_eq!(HttpMethod::Get.to_string(), "GET");

// Retry decisions without a hand-written table.
assert!(HttpMethod::Put.is_idempotent());
assert!(!HttpMethod::Post.is_idempotent());

// The vocabulary an `http_call` task may name, for validating your own
// operator-facing allow-lists against something the compiler keeps honest.
assert_eq!(HttpMethod::ALL.len(), 5);
}

HttpMethod::ALL is scoped to what http_call accepts. It is deliberately not a general list of HTTP methods — don’t reuse it to validate inbound routes, which may legitimately accept HEAD or OPTIONS.

path_logic is an alias for path, not a second field — supplying both is a duplicate field error. Same for body_logic / body.

enrich

Fetch external data and merge it into the message context at a specified path. A specialization of http_call aimed at the “look up and attach” pattern.

Configuration

{
    "function": {
        "name": "enrich",
        "input": {
            "connector": "customer_lookup",
            "method": "GET",
            "path": { "cat": ["/customers/", {"var": "data.customer_id"}] },
            "merge_path": "data.customer",
            "timeout_ms": 5000,
            "on_error": "skip"
        }
    }
}

Parameters

on_error: skip is useful when enrichment is best-effort and an absent upstream service shouldn’t fail the workflow.

publish_kafka

Emit the message (or a derived value) to a Kafka topic.

Configuration

{
    "function": {
        "name": "publish_kafka",
        "input": {
            "connector": "events_cluster",
            "topic": { "cat": ["orders.", {"var": "data.region"}] },
            "key": { "var": "data.order_id" },
            "value": { "var": "data" }
        }
    }
}

Parameters

The handler decides exactly how to render the produced value — for example, sending the entire message JSON when value is omitted.

Connectors

The connector field is a string that your handler resolves into a concrete client (HTTP client + base URL, Kafka producer + cluster config, …). The engine does not interpret it. A typical layout:

struct HttpCallHandler {
    connectors: HashMap<String, HttpConnector>,  // "user_service" -> &Client + base_url
}

This separation keeps secrets out of workflow JSON and lets you swap endpoints (staging / prod) without touching rule definitions.

When a request does need a per-workflow credential — a bearer token, a partner key in the body, a signed path — read it with {"secret": "name"} inside the relevant parameter rather than seeding it into metadata, which every trace snapshot would then carry. Since 3.9 that includes headers, which is usually where it belongs:

"headers": { "Authorization": { "cat": ["Bearer ", {"secret": "partner_token"}] } }

The value comes from the engine’s secret store and is never recorded; see Secrets.

Why typed configs matter

Compared to free-form Custom configs:

• Startup-time validation — bad config fails at Engine::new()
• Pre-compiled JSONLogic — path_logic, body_logic, key_logic, value_logic are all compiled once; the handler reads Arc<Logic> from the config and evaluates at zero allocation cost in the hot path
• Stable shape — the same config struct is shared by every handler in the ecosystem, so handlers from different crates can be swapped without rewriting workflows