Struct ChainCtx 

#[non_exhaustive]
pub struct ChainCtx {
    pub thread: usize,
    pub queues: ShardQueues,
    pub budget: Arc<InflightBudget>,
    pub pipeline: String,
    pub source_framing: FramingContract,
    /* private fields */
}

Per-thread wiring handed to the chain factory, carrying everything the terminal .sink(...) stage needs, so assemblies stop threading queues, budget, and the pipeline name by hand.

Passed by value, once per pipeline thread; move the fields into the chain being built. Not Clone (see the module docs on drop ordering).

Fields (Non-exhaustive)

Non-exhaustive structs could have additional fields added in future. Therefore, non-exhaustive structs cannot be constructed in external crates using the traditional Struct { .. } syntax; cannot be matched against without a wildcard ..; and struct update syntax will not work.

thread: usize

Zero-based pipeline thread index.

queues: ShardQueues

This thread’s clone of the shard-queue senders for the first installed sink, the back-compat handle for single-sink pipelines (.sink(...)). Multi-sink pipelines resolve each branch’s queues by name via sink instead.

budget: Arc<InflightBudget>

The shared in-flight byte budget.

pipeline: String

The pipeline name (ChainBuilder::with_metrics’s first argument).

source_framing: FramingContract

How the source frames its payloads (Source::framing_contract). Hand it to a deserializer builder (e.g. JsonDeserializerBuilder:: for_source_framing) so the deserializer’s granularity is derived from the source and a double-framing configuration is rejected.

Implementations

impl ChainCtx

pub fn sink(&self, name: &str) -> SinkCtx

The named sink’s handles (name, shard queues, shared in-flight budget) for a split-terminal branch. Pass the result straight to SplitBuilder::add. The single-sink .sink() sugar installs its sink under the name "default".

Panics

Panics if no sink was installed under name. The chain factory runs once per thread, on the cold path, before any data flows.

pub fn chunk(&self) -> ChunkConfig

The first installed sink’s resolved terminal-stage chunking, the single-sink counterpart to queues, fed straight to the chain’s .sink(...) terminal:

let chunk_cfg = ctx.chunk(); // bind before `with_metrics` moves `ctx.pipeline`
let orders = chain(BytesPassthrough)
    .with_metrics(ctx.pipeline, "orders")
    .sink(encoder, KeyHashRouter, chunk_cfg, ctx.queues, ctx.budget)
    .build();

It resolves the per-sink YAML chunk: block (or SinkOptions::with_chunk, or the 64 KiB default) at assembly time. Split pipelines take each branch’s chunk from sink instead.

pub fn meter( &self, component: impl Into<SharedString>, component_type: impl Into<SharedString>, ) -> Meter

A Meter for a pipeline author’s own metrics, pre-labeled with the pipeline name plus the component / component_type you name and scoped to the spate_custom_ namespace. Resolve handles from it once here (the factory runs once per thread, before data flows) and move them into the operator closures that touch them:

// Pass the LOCAL name — this registers `spate_custom_enrich_hits_total`.
let hits = ctx.meter("enrich", "map").counter("enrich_hits_total", &[]);
// ... move `hits` into a `.inspect(move |r| { hits.increment(1); })`

The resulting series carry pipeline/component/component_type like every framework series and live under the spate_ umbrella, so they join cleanly in a query. You pass local names; the Meter adds the spate_custom_ prefix. The metrics reference carries the umbrella and the label set in full.

Trait Implementations

impl Debug for ChainCtx

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

Formats the value using the given formatter.