Struct WorkBatch 

pub struct WorkBatch<T: CommitToken> {
    pub records: Vec<Record>,
    pub commit_tokens: Vec<T>,
    pub dlq_entries: Vec<FilteredDlqEntry>,
}

Available on crate feature transport only.

The canonical zero-copy block of work records.

One WorkBatch is the single currency through get -> process -> send -> commit. The commit_tokens are the INPUT source acks – fired once after the WHOLE block is sent (at-least-once). T generalises the source ack (Kafka offset, HTTP responder, fetch cursor, …).

commit_tokens.len() is intentionally decoupled from records.len(): a fan-out transform may grow or shrink the record count while the commit tokens stay equal to the input acks.

Fields

records: Vec<Record>

The work records in this block.

commit_tokens: Vec<T>

Source acks for the whole block. Fired after the block is fully sent. Length is NOT tied to records.len().

dlq_entries: Vec<FilteredDlqEntry>

Inline-DLQ entries carried forward (no-silent-drop contract).

Implementations

impl<T: CommitToken> WorkBatch<T>

pub fn empty() -> Self

An empty batch (no records, no tokens, no DLQ entries).

pub fn from_records(records: Vec<Record>) -> Self

A batch of records with no commit tokens and no DLQ entries.

Useful for records that were generated downstream (e.g. a transform that emits new records) and do not themselves carry a source ack.

pub fn new(records: Vec<Record>, commit_tokens: Vec<T>) -> Self

A batch of records plus their source commit tokens (no DLQ entries).

pub fn with_dlq_entries(self, dlq_entries: Vec<FilteredDlqEntry>) -> Self

Attach inline-DLQ entries to this batch, consuming and returning it.

pub fn is_empty(&self) -> bool

Whether the batch has no records (DLQ entries may still be present).

pub fn len(&self) -> usize

Number of records.

pub fn record_count(&self) -> usize

Number of records (explicit alias for WorkBatch::len).

pub fn total_payload_bytes(&self) -> usize

Total payload bytes across all records.

Needed by the byte-budget governor to size a block against a memory budget. – Sums with saturating addition, so a pathological block can never overflow usize; it clamps at usize::MAX instead of wrapping.

pub fn map_records(self, f: impl FnOnce(Vec<Record>) -> Vec<Record>) -> Self

Transform the records while PRESERVING the commit tokens and DLQ entries.

This is the safe shape for the future process handler: a transform may grow, shrink, or rewrite the records (fan-out / fan-in), but the source acks and inline-DLQ entries flow through untouched.

pub fn single(blob: Bytes) -> Self

One record holding the WHOLE blob (no framing).

The single record’s payload is the blob itself (zero-copy move of the Bytes handle), key is None, headers empty, and the metadata format is auto-detected via PayloadFormat::detect.

pub fn from_ndjson(blob: Bytes) -> Self

Frame an NDJSON (newline-delimited JSON) blob into one record per line.

Splits on \n using memchr. Each line is trimmed of a trailing \r (so Windows CRLF endings work) and of surrounding ASCII whitespace; blank or whitespace-only lines are skipped. Each surviving line becomes a Record whose payload is a zero-copy blob.slice(start..end) view. format is set to PayloadFormat::Json.

This is framing, not parsing – it never fails. A line that is not valid JSON still becomes a record; the downstream parser surfaces that.

pub fn from_json_array(blob: Bytes) -> Result<Self, FramingError>

Frame a top-level JSON array blob into one record per top-level element.

Each element becomes a zero-copy blob.slice(start..end) view; element VALUES are not parsed. The byte-level scanner tracks string state (so [, ], {, } and , inside a string are literal), JSON escapes (\", \\), and container depth, so nested objects/arrays frame correctly. An empty array [] yields zero records (Ok).

Errors

Returns FramingError when the blob is not a well-framed top-level array: missing opening [, unterminated array/string, an empty element position (leading / trailing / doubled comma), an over-close, or trailing non-whitespace after the closing ].

Trait Implementations

impl<T: Debug + CommitToken> Debug for WorkBatch<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T: CommitToken> From<Message<T>> for WorkBatch<T>

fn from(msg: Message<T>) -> Self

Collapse a single crate::Message into a one-record WorkBatch.

The message payload is already Bytes – this is a move, not a copy.

impl<T: CommitToken> From<RecvBatch<T>> for WorkBatch<T>

fn from(batch: RecvBatch<T>) -> Self

Collapse a RecvBatch into a WorkBatch.

Each Message<T> becomes a Record; its token is collected into commit_tokens (preserving order); dlq_entries carry straight across. Payloads are already Bytes – each is a move, not a copy.