Struct ChunkStep 

pub struct ChunkStep<I, O, R, P, W> { /* private fields */ }

A validated one-step chunk definition.

Implementations

impl<I, O, R, P, W> ChunkStep<I, O, R, P, W>

where R: ItemReader<I>, P: ItemProcessor<I, O>, W: ItemWriter<O>,

pub fn new( name: StepName, size: ChunkSize, reader: R, processor: P, writer: W, transactions: Arc<dyn ChunkTransactionManager>, completion: Arc<dyn ChunkCompletion>, ) -> Self

Constructs a chunk step from facade-owned component and transaction ports.

pub fn with_completion(self, completion: Arc<dyn ChunkCompletion>) -> Self

Replaces the post-commit completion observer installed by Self::new.

Additive parity with the other post-construction with_* methods: Self::new requires a ChunkCompletion up front because it is a required construction port, not because every caller has an opinion about it. crate::ChunkPipelineBuilder installs a no-op default and uses this method to let a caller override it without hand-rolling a ChunkCompletion implementation for the common case that does not need one.

pub fn with_completion_policy(self, policy: Arc<dyn CompletionPolicy>) -> Self

Installs an additional early-completion policy.

The configured ChunkSize passed to Self::new always remains the hard resource-safety ceiling for buffered items; installing a policy here only allows a chunk to stop earlier than that ceiling, per REPEAT-POLICY-001. Without a policy, completion is exactly the pre-#151 ChunkSize-only behavior.

Also registers every ItemStream that policy (or, for a crate::CompositeCompletionPolicy, any of its members at any nesting depth) reports through CompletionPolicy::stream_registrations, bound to that exact instance. This is the only place that registration happens, so an AdaptiveCompletionPolicy gets its persisted decision wired correctly whether it is installed directly here or nested anywhere inside a composite – there is no second, composite-blind path that could silently drop it.

Calling this again to replace a previously installed policy first removes exactly the previous policy’s own registered streams – every entry this method itself tagged as policy-owned, never a manually registered Self::with_item_stream entry that happens to coexist, even under the identical ComponentStreamIdentity: a policy replacement is a full replacement of what this method installed, not an accumulation of every policy this step ever installed, and ownership is tracked per entry rather than inferred from identity equality, so a coincidental identity collision with a manual registration can never cause that manual entry to be removed.

pub fn with_adaptive_completion_policy( self, policy: Arc<AdaptiveCompletionPolicy>, ) -> Self

Registers one AdaptiveCompletionPolicy instance as this step’s completion policy.

A thin, discoverable alias for Self::with_completion_policy: CompletionPolicy::stream_registrations already binds this exact instance’s persisted decision to the step, so there is exactly one registration mechanism regardless of which method installs it – never a second path that could drift onto a different instance.

pub fn with_chunk_listener(self, listener: Arc<dyn ChunkListener>) -> Self

Registers a chunk listener in deterministic before-order.

pub fn with_item_listeners(self, listeners: ItemListenerSet<I, O>) -> Self

Installs the authoritative item, retry, and skip listener families.

The set replaces any previously installed families.

pub fn with_fault_runtime(self, fault: FaultRuntime) -> Self

Installs bounded retry, backoff, skip, and rollback behavior.

Without a fault runtime every component failure fails the step after a known rollback, which is the M2 behavior.

pub fn with_listener(self, listener: Arc<dyn StepExecutionListener>) -> Self

Registers a step listener in deterministic before-order.

pub fn with_item_stream( self, identity: ComponentStreamIdentity, stream: impl ItemStream + 'static, contract: StreamStateContract, ) -> Self

Registers one namespaced ItemStream in deterministic registration order.

identity must match the namespace registered for this stream in the ChunkComponentRevisions this step’s definition declares (see ChunkComponentRevisions::with_stream_revision): it is both the restart-relevant fingerprint token and the runtime key used to associate this stream with its committed durable state. ChunkJob::new/FlowJob::with_chunk_step reject a definition where the runtime registrations and declared revisions are not the same set.

contract binds this stream’s expected schema/codec identity, version, and restartability declaration so the runtime can validate and migrate an inherited envelope – and enforce restartability at binding time – before open ever runs, without introspecting this stream’s opaque internal state.

pub const fn name(&self) -> &StepName

Borrows the step name.

pub const fn size(&self) -> ChunkSize

Returns the configured chunk size.

pub async fn execute( &mut self, correlation: &ExecutionCorrelation, stop: &StopToken, ) -> ChunkExecutionReport

where I: Send + Sync, O: Send + Sync,

Executes this step deterministically on the caller’s async runtime.

The reader is stateful, so the definition is mutably borrowed for the duration of the run. Only counts returned by successful transaction commits appear in the report. correlation identifies the execution for item, retry, and skip listeners; it never reaches a component.

Trait Implementations

impl<I, O, R, P, W> Debug for ChunkStep<I, O, R, P, W>

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

Formats the value using the given formatter.