Struct BatchTransform 

pub struct BatchTransform { /* private fields */ }

Groups envelopes into batches by maximum count and/or maximum time window.

When either limit is reached, the batch is emitted as a single envelope whose payload contains an array of the original payloads under the configured payload_key.

Meta derivation

The emitted envelope inherits meta from the first envelope in the batch. Its key is set to batch-{first_timestamp_ms} so downstream sinks that require a key have a deterministic value.

Acknowledgement

Because channels are the acknowledgement boundary, the batched envelope is acknowledged only when downstream work for it completes. This in turn acknowledges every constituent envelope, since none are released until the batch envelope is delivered to the next stage.

Cancellation

By default (flush_on_cancel = true), any partial batch is flushed immediately when the cancellation token fires. When set to false, partial batches are dropped on cancellation.

Implementations

impl BatchTransform

pub fn new( id: impl Into<String>, max_size: usize, max_delay_ms: Option<u64>, payload_key: impl Into<String>, flush_on_cancel: bool, ) -> Self

Trait Implementations

impl Transform for BatchTransform

fn id(&self) -> &str

fn set_node_ctx(&mut self, ctx: NodeCtx)

Attach the per-node observability context. Called by spawn_pipeline after the transform is built but before it runs. Default no-op — full-control transforms that want metrics override this and store the ctx; BasicTransform already does so for the common path.

fn run<'async_trait>( self: Box<Self>, rx: Receiver<Envelope>, tx: Sender<Envelope>, cancel: CancellationToken, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait,