Buffers & Rate

Datum is backpressure-first: producers slow down when consumers can't keep up (see Backpressure). The operators on this page override that default when you want to trade off latency, throughput, or memory.

buffer

buffer(n, strategy) inserts an N-element queue between producer and consumer. When the queue is full, strategy determines what happens:

use datum::{Flow, OverflowStrategy, Sink, Source};

// buffer(n, strategy) inserts an n-element buffer between producer and consumer.
// Backpressure holds the producer when the buffer is full.
let items: Vec<u64> = Source::from_iter(1_u64..=5)
    .via(Flow::identity().buffer(4, OverflowStrategy::Backpressure))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

OverflowStrategy variants:

Variant	On overflow
Backpressure	Slow the producer (default stream behavior without a buffer)
DropNew	Discard the incoming element
DropHead	Discard the oldest element in the buffer
DropTail	Discard the newest element already in the buffer
DropBuffer	Discard the entire buffer contents
Fail	Fail the stream with StreamError::Failed("Buffer overflow (max capacity was: N)")

detach() is a shorthand for buffer(1, OverflowStrategy::Backpressure). It breaks a fused chain at one point, allowing the two sides to run on separate scheduling ticks.

throttle

throttle(elements, per, maximum_burst, mode) limits throughput using a token bucket. At most elements tokens refill every per duration; maximum_burst sets the initial bucket level (and the ceiling for accumulated tokens):

use datum::{Flow, Sink, Source, ThrottleMode};
use std::time::Duration;

// throttle(1, per, burst, Shaping) limits throughput to 1 element per `per`
// while spacing them evenly; Enforcing would drop excess elements instead.
let items: Vec<u64> = Source::from_iter(1_u64..=3)
    .via(Flow::identity().throttle(1, Duration::from_micros(500), 1, ThrottleMode::Shaping))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

ThrottleMode values:

Mode	Behavior when the bucket is empty
Shaping	Hold the element until enough tokens accumulate
Enforcing	Fail the stream with StreamError::Failed("Maximum throttle throughput exceeded.")

throttle_with_cost(cost, per, burst, cost_fn, mode) lets each element carry a variable cost instead of counting every element as 1.

delay, delay_with, and initial_delay

delay(duration, strategy) time-shifts each element by holding it in a bounded delay buffer before it is visible downstream:

use datum::{DelayOverflowStrategy, Sink, Source};
use std::time::{Duration, Instant};

// delay(duration, strategy) holds each element until its delay expires.
// Backpressure keeps upstream from outrunning the bounded delay buffer.
let started = Instant::now();
let items: Vec<&str> = Source::from_iter(["one", "two"])
    .delay(
        Duration::from_millis(20),
        DelayOverflowStrategy::Backpressure,
    )
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

assert_eq!(items, vec!["one", "two"]);
assert!(started.elapsed() >= Duration::from_millis(10));

Datum's delay buffer holds 16 delayed elements. If upstream produces the 17th delayed element before downstream has released any of the pending elements, DelayOverflowStrategy decides how to make room:

Strategy	On overflow
EmitEarly	Release the oldest delayed element before its deadline
DropHead	Drop the oldest delayed element
DropTail	Drop the newest delayed element already in the buffer
DropBuffer	Clear all delayed elements
DropNew	Drop the incoming element
Backpressure	Hold upstream until space is available
Fail	Fail the stream with a delay-buffer overflow error

use datum::{DelayOverflowStrategy, Source, StreamError};
use std::time::Duration;

// The delay buffer holds 16 delayed elements. Fail turns the 17th pending
// element into a stream failure instead of dropping or releasing an item.
let result = Source::from_iter(0_u64..17)
    .delay(Duration::from_secs(60), DelayOverflowStrategy::Fail)
    .run_collect();

assert!(matches!(
    result,
    Err(StreamError::Failed(message))
        if message.contains("Buffer overflow for delay operator")
));

Use delay_with(supplier, strategy) when each element should choose its own delay. The supplier is called once per materialization and returns the FnMut(&Out) -> Duration used for that run:

use datum::{DelayOverflowStrategy, Sink, Source};
use std::time::Duration;

// delay_with builds a per-materialization delay function.
let items: Vec<&str> = Source::from_iter([("fast", 0_u64), ("slow", 15)])
    .delay_with(
        || |reading: &(&str, u64)| Duration::from_millis(reading.1),
        DelayOverflowStrategy::Backpressure,
    )
    .map(|(name, _delay_ms)| name)
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

assert_eq!(items, vec!["fast", "slow"]);

initial_delay(duration) delays only the beginning of the stream. After the first pull is allowed, the rest of the stream proceeds normally:

use datum::{Sink, Source};
use std::time::{Duration, Instant};

// initial_delay delays the first pull, then lets the stream run normally.
let started = Instant::now();
let item = Source::single("ready")
    .initial_delay(Duration::from_millis(20))
    .run_with(Sink::head())
    .unwrap()
    .wait()
    .unwrap();

assert_eq!(item, "ready");
assert!(started.elapsed() >= Duration::from_millis(10));

conflate

conflate(f) lets a fast producer merge elements into a single downstream value when the consumer is not ready. The function f(accumulated, next) is called each time a new element arrives while the consumer is busy:

use datum::{Flow, Sink, Source};

// conflate(f) merges buffered elements when the consumer is slower than
// the producer. In a synchronous fused chain the consumer keeps up, so
// all items pass through and the sum across them is always correct.
let items: Vec<u64> = Source::from_iter(1_u64..=4)
    .via(Flow::identity().conflate(|acc, x| acc + x))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

conflate_with_seed(seed, aggregate) provides separate functions for creating the initial accumulator (seed(first_element)) and extending it (aggregate(acc, next_element)).

In a synchronous fused chain where producer and consumer run at the same speed, conflate does not coalesce: elements pass through one at a time. Coalescing only occurs when the consumer is genuinely slower than the producer (e.g. there is an async boundary or a slow map downstream).

batch and batch_weighted

batch(max, seed, aggregate) groups up to max elements into one aggregate. Unlike conflate, batch has a cap: once the aggregate's "weight" reaches max, it is emitted and a new one starts:

use datum::{Flow, Sink, Source};

// batch(max, seed, aggregate) groups up to `max` elements into one aggregate value.
// seed(first_element) starts the batch; aggregate(batch, next) extends it.
// Like conflate, batching only coalesces when the producer outruns the consumer.
let items: Vec<Vec<u64>> = Source::from_iter(1_u64..=4)
    .via(Flow::identity().batch(
        8,
        |x| vec![x],
        |mut agg, x| {
            agg.push(x);
            agg
        },
    ))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

batch_weighted(max, cost_fn, seed, aggregate) uses a per-element cost function to determine when the batch is full.

aggregate_with_boundary

aggregate_with_boundary(allocate, aggregate, harvest, emit_on_timer) builds one aggregate at a time and emits it when either the aggregate closure reports a boundary or the optional AggregateTimer predicate fires.

• allocate() creates an empty aggregate.
• aggregate(agg, item) returns (updated_agg, ready). Set ready to true to emit immediately.
• harvest(agg) converts the aggregate into the downstream value.
• emit_on_timer can be Some(AggregateTimer::new(predicate, interval)) to flush a non-empty aggregate after a time window.

This example emits a sensor batch when it reaches three readings, or earlier if a timer sees a non-empty partial batch during a quiet period:

use datum::{AggregateTimer, Sink, Source};
use std::time::Duration;

#[derive(Clone, Debug)]
struct SensorReading {
    sensor: &'static str,
    value: u64,
}

#[derive(Debug, PartialEq, Eq)]
struct SensorBatch {
    sensors: Vec<&'static str>,
    total: u64,
}

let first_window = Source::from_iter([
    SensorReading {
        sensor: "thermostat",
        value: 21,
    },
    SensorReading {
        sensor: "humidity",
        value: 44,
    },
]);
let next_window = Source::from_iter([
    SensorReading {
        sensor: "thermostat",
        value: 22,
    },
    SensorReading {
        sensor: "humidity",
        value: 45,
    },
    SensorReading {
        sensor: "pressure",
        value: 101,
    },
])
.initial_delay(Duration::from_millis(60));

let batches: Vec<SensorBatch> = first_window
    .concat(next_window)
    .aggregate_with_boundary(
        Vec::<SensorReading>::new,
        |mut batch, reading| {
            batch.push(reading);
            let reached_size_boundary = batch.len() >= 3;
            (batch, reached_size_boundary)
        },
        |batch| SensorBatch {
            sensors: batch.iter().map(|reading| reading.sensor).collect(),
            total: batch.iter().map(|reading| reading.value).sum(),
        },
        Some(AggregateTimer::new(
            |batch: &Vec<SensorReading>| !batch.is_empty(),
            Duration::from_millis(10),
        )),
    )
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

assert_eq!(
    batches,
    vec![
        SensorBatch {
            sensors: vec!["thermostat", "humidity"],
            total: 65,
        },
        SensorBatch {
            sensors: vec!["thermostat", "humidity", "pressure"],
            total: 168,
        },
    ]
);

expand and extrapolate

expand(f) is the mirror of conflate: it lets the downstream pull faster than the upstream produces. When downstream demands an element but the upstream has nothing ready, expand calls f on the last-seen element and emits from the resulting iterator until the upstream supplies a new element:

use datum::{Flow, Sink, Source};
use std::time::Duration;

let first = Source::single(String::from("steady"));
let delayed_next =
    Source::single(String::from("next")).initial_delay(Duration::from_millis(50));

// Repeat the last element up to three times while upstream is slow.
let readings: Vec<String> = first
    .concat(delayed_next)
    .via(
        Flow::identity()
            .expand(|reading: String| std::iter::repeat_with(move || reading.clone()).take(3)),
    )
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

assert_eq!(readings, ["steady", "steady", "steady", "next"]);

extrapolate(f, initial) is a variant that requires Out: Clone and uses an optional initial value for the very first upstream pull. It is typically used to extrapolate a sensor reading while waiting for the next measurement.

Both operators have rate-dependent behavior: in a synchronous fused chain where the upstream always has an element ready, elements pass through without expansion.

Summary

Operator	Direction	What it trades
buffer(n, strategy)	Producer → Consumer	Memory for smoothed throughput
throttle(n, per, burst, mode)	Producer → Consumer	Throughput for latency control
delay(duration, strategy)	Producer → Consumer	Time shifting for bounded buffering
initial_delay(duration)	Producer → Consumer	Delayed startup
conflate(f)	Consumer ← Producer	Memory for producer speed
expand(f)	Consumer → Producer	Consumer speed for upstream gaps
batch(max, seed, agg)	Consumer ← Producer	Memory (bounded) for reduced message count
aggregate_with_boundary(...)	Consumer ← Producer	Window boundaries for reduced message count
detach()	Both	One decoupling point

Next steps

• Backpressure — the default flow-control model
• Error Handling — what happens when a stream fails