Async, sync, and Kafka — tflo's three execution modes

You develop against a CSV file. You test against an async stream. You deploy against Kafka. Same graph. Three execution modes.

This is the core design rule of tflo: the graph doesn’t know where data comes from. You define the computation once — the operators, the signal detectors, the crossings — and the execution mode is a separate concern. Swap the input, keep the graph.

Sync — iterators

For development, testing, and batch processing, use the iterator API. It’s the simplest path from data to results.

use tflo_core::prelude::*;
use tflo_ops::prelude::*;

let results: Vec<f64> = ticks.into_iter()
    .tflo(|t| {
        t.timestamp(|x| x.ts);
        let price = t.prop(|x| x.price);
        price.sma(20_u64.secs())
    })
    .collect();

tflo() strips away the input record and gives you just the computed values. If you need the original record alongside the result — for debugging, enrichment, or routing — use .with():

let results: Vec<(Tick, f64)> = ticks.into_iter()
    .with(|t| {
        t.timestamp(|x| x.ts);
        let price = t.prop(|x| x.price);
        price.sma(20_u64.secs())
    })
    .collect();

The closure receives a TFlowBuilder and returns a compilation unit (Comp). Same pattern whether you’re computing one indicator or ten.

Async — streams

When data arrives asynchronously — WebSocket feeds, gRPC streams, Kafka consumer callbacks — use the stream API enabled by the async feature on tflo-core.

[dependencies]
tflo-core = { version = "0.0.2", features = ["async"] }

With the feature enabled, TFloStreamExt is available from tflo_core::prelude::*:

use tflo_core::prelude::*;
use tflo_ops::prelude::*;
use tokio_stream::StreamExt;

let mut stream = tick_stream
    .tflo(|t| {
        t.timestamp(|x| x.ts);
        let price = t.prop(|x| x.price);
        price.sma(20_u64.secs())
    });

while let Some(sma) = stream.next().await {
    println!("SMA: {sma}");
}

The same TFlowBuilder closure. The same Comp chain. But now it’s driven by Stream::poll_next instead of Iterator::next. The TFloStream type wraps the underlying stream and the compiled graph, emitting values as they arrive.

There’s also tflo_with() for streams (note the tflo_ prefix on the async variant), returning TFlowWithStream<S, R, O> which yields (R, O) tuples — original record plus computed value.

See tflo_core::r#async for the TFloStreamExt trait. Enable with tflo-core = { features = ["async"] }.

Keyed — partitioned state

When you have multiple entities sharing a graph definition — one running mean per sensor, one signal tracker per RF emitter — use keyed execution. The graph structure is the same for every key, but the state is isolated:

use tflo_core::keyed::OutOfOrderPolicy;

let results: Vec<_> = detections.into_iter()
    .tflo_keyed(
        |r| r.sensor_id.clone(),                  // key function
        OutOfOrderPolicy::Buffer { max_lateness_ms: 5000 },  // late data
        |t| {
            t.timestamp(|x| x.ts);
            let snr = t.prop(|x| x.snr);
            snr.sma(5_u64.secs())
        }
    )
    .collect();

Every record from sensor “north-3” routes to its own graph instance. Same for “north-4”, “gate-1”, and the rest. The OutOfOrderPolicy handles records that arrive out of timestamp order — error, drop, or buffer with a lateness window.

Keyed execution also works on streams via the async feature.

Kafka — production

The tflo-connect-kafka crate defines the integration pattern for Kafka. The KafkaTfloAdapter wraps keyed execution around a Kafka consumer — partitions map to keys, checkpointing coordinates state snapshots with offset commits.

The architecture pattern works in four layers:

1. Extract key and timestamp per record — the key function and timestamp function isolate identity and ordering from the transport.
2. Route to per-key graph instances via tflo_keyed — each key gets its own graph, with its own state, its own windows.
3. Coordinate state snapshots + offset commits on checkpoint boundaries — the checkpoint policy decides when to persist, and the state store + cursor store handle the actual I/O.
4. Persist to state store (S3, files, etc.) — snapshots land in whatever backend you wire in, making the pipeline restartable from any committed offset.

The crate is a reference implementation showing the integration pattern without depending on a specific Kafka client library. For production Kafka integration, wire in a Kafka client library (e.g., rdkafka) and follow the adapter pattern shown in the crate.

The pattern is always the same

Every execution mode follows this blueprint:

1. Create a TFlowBuilder — the closure receives one.
2. Set the timestamp — via t.timestamp(...) for time-based windows.
3. Define computations — chain Comp methods: prop(...), .sma(), .cross_above().
4. Execute — the mode-specific wrapper (iterator, stream, keyed, Kafka) handles the rest.

The graph doesn’t know it’s being driven by an iterator vs. a stream vs. Kafka. It calls step on each node, advances state, returns output. Your code is the same in all three cases.

Develop against a recorded CSV. Test against a replay stream. Run against Kafka. One graph.

Source code

The full working example is available on GitHub: tflo-examples/examples/three-execution-modes

— Matt