StreamSpec

Overview

The io.flowwarden.stream.registration package (since 1.0.0-rc.6) declares Change Streams without annotations. A StreamDefinitionContributor bean is called once at bootstrap with a StreamRegistration; each registration.stream(name, documentType) call opens a StreamSpec.Builder. The resulting StreamSpec is immutable and is converted internally to the same definition an annotated class produces. For the motivation and worked examples, see the Programmatic Registration guide.

Bootstrap-only. Contributors run after all singleton beans are created and before any stream starts. There is no API to add or remove a stream on a running instance.

`StreamDefinitionContributor`

@FunctionalInterface
public interface StreamDefinitionContributor {
    void contribute(StreamRegistration registration);
}

Discovered as Spring beans. Every contributor’s contribute runs exactly once. A duplicate stream name across contributors or against an annotated stream fails startup.

`StreamRegistration`

Method	Description
`<T> StreamSpec.Builder<T> stream(String name, Class<T> documentType)`	Opens a builder for one stream. The builder is collected automatically — do not call `build()` yourself.

`StreamSpec.Builder<T>`

Target and behaviour

Method	Annotation equivalent	Default
`collection(String)`	`@ChangeStream.collection`	resolved from `@Document` on `documentType`
`database(String)`	`@ChangeStream.database`	template’s default database
`mongoTemplateRef(String)`	`@ChangeStream.mongoTemplateRef`	primary template
`enabled(boolean)`	`@ChangeStream.enabled`	`true`
`autoStart(boolean)`	`@ChangeStream.autoStart`	`true`
`fullDocument(FullDocumentMode)`	`@ChangeStream.fullDocument`	same as annotation
`fullDocumentBeforeChange(FullDocumentBeforeChangeMode)`	`@ChangeStream.fullDocumentBeforeChange`	same as annotation
`deploymentMode(DeploymentMode)`	`@ChangeStream.deploymentMode`	same as annotation

Handlers

One handler per operation type; a second registration for the same operation throws IllegalStateException.

Method	Handler shape
`onInsert(ContextHandler<T>)` / `onInsert(DocumentHandler<T>)`	`void handle(ChangeStreamContext<?> ctx)` / `void handle(T document, ChangeStreamContext<?> ctx)`
`onUpdate(...)`, `onDelete(...)`, `onReplace(...)`	same two shapes
`onChange(ContextHandler<T>)`	catch-all, context only
`onInsertReactive(ReactiveContextHandler<T>)` / `onInsertReactive(ReactiveDocumentHandler<T>)`	`Mono<Void> handle(...)`
`onUpdateReactive(...)`, `onDeleteReactive(...)`, `onReplaceReactive(...)`, `onChangeReactive(...)`	reactive variants

The handler interfaces live in io.flowwarden.stream.core. Imperative handlers on a reactive stream (or the reverse) are rejected at bootstrap.

Pipeline, filter, errors

Method	Annotation equivalent	Rules
`pipeline(Supplier<List<Bson>>)`	`@Pipeline`	Supplier evaluated once at stream start. One per stream. A supplier that throws or returns `null` fails the stream start with `IllegalStateException`.
`filter(Predicate<ChangeStreamContext<T>>)`	`@Filter`	Evaluated on every event, before dispatch. One per stream. Rejected at bootstrap when combined with a typed handler on DELETE/DROP/INVALIDATE.
`onError(ErrorHandler, Class<? extends Throwable>...)`	`@OnError`	Repeatable. No types = catch-all (at most one per stream). Each exception type may be claimed by one handler per stream.

ErrorHandler (in io.flowwarden.stream.core):

@FunctionalInterface
public interface ErrorHandler {
    ErrorAction handle(Throwable ex, ChangeStreamContext<?> ctx);
}

Resolution and ErrorAction semantics are those of @OnError. A handler that throws falls back to RETHROW.

Policies

Method	Annotation equivalent
`checkpoint(CheckpointSpec)`	`@Checkpoint`
`retryPolicy(RetryPolicySpec)`	`@RetryPolicy`
`deadLetterQueue(DeadLetterQueueSpec)`	`@DeadLetterQueue`
`mongoDlqOptions(MongoDlqOptionsSpec)`	`@MongoDlqOptions`

Spec types

Each spec type is an immutable value with a builder() and a defaults() factory. defaults() returns exactly the annotation’s default attribute values.

CheckpointSpec

Builder method	Annotation attribute
`saveEveryN(int)`	`@Checkpoint.saveEveryN`
`saveIntervalSeconds(int)`	`@Checkpoint.saveIntervalSeconds`
`idleHeartbeatIntervalSeconds(int)`	`@Checkpoint.idleHeartbeatIntervalSeconds`
`startPosition(StartPosition)`	`@Checkpoint.startPosition`
`onHistoryLost(OnHistoryLost)`	`@Checkpoint.onHistoryLost`

See @Checkpoint for the semantics and bounds.

RetryPolicySpec

Builder method	Annotation attribute
`maxAttempts(int)`	`@RetryPolicy.maxAttempts`
`initialDelay(String)`	`@RetryPolicy.initialDelay`
`maxDelay(String)`	`@RetryPolicy.maxDelay`
`multiplier(double)`	`@RetryPolicy.multiplier`
`retryOn(List<Class<? extends Throwable>>)`	`@RetryPolicy.retryOn`
`noRetryOn(List<Class<? extends Throwable>>)`	`@RetryPolicy.noRetryOn`
`jitter(boolean)`	`@RetryPolicy.jitter`

See @RetryPolicy.

DeadLetterQueueSpec

Builder method	Annotation attribute
`enabled(boolean)`	`@DeadLetterQueue.enabled`
`retentionDays(int)`	`@DeadLetterQueue.retentionDays`
`includeOriginalDocument(boolean)`	`@DeadLetterQueue.includeOriginalDocument`
`includeStackTrace(boolean)`	`@DeadLetterQueue.includeStackTrace`

See @DeadLetterQueue.

MongoDlqOptionsSpec

Builder method	Annotation attribute
`collection(String)`	`@MongoDlqOptions.collection`

`StreamSpec<T>` accessors

The built spec exposes what was declared, read-only: name(), documentType(), collection(), database(), enabled(), autoStart(), fullDocument(), fullDocumentBeforeChange(), deploymentMode(), mongoTemplateRef(), checkpoint(), retryPolicy(), deadLetterQueue(), mongoDlqOptions(), handler(OperationType), typedHandlers(), onChangeHandler(), pipeline(), filter(), errorHandlers(). Optional-valued where the annotation attribute is optional; collections are defensive copies.

Not covered

zone — no builder method yet. A stream needing a zone stays annotated.
Hot registration — contributions are fixed for the lifetime of the context.

Validation

Contributed streams go through the validator shared with the annotation path. Any violation — checkpoint/retry/DLQ bounds, collection resolution, mongoTemplateRef type, handler mode, filter compatibility, duplicate onError claims, duplicate stream name — fails application startup with the same messages an annotated class would produce.

Annotations

Registration API

Runtime

SPI

Overview

`StreamDefinitionContributor`

`StreamRegistration`

`StreamSpec.Builder<T>`

Target and behaviour

Handlers

Pipeline, filter, errors

Policies

Spec types

`StreamSpec<T>` accessors

Not covered

Validation

​Overview

​StreamDefinitionContributor

​StreamRegistration

​StreamSpec.Builder<T>

​Target and behaviour

​Handlers

​Pipeline, filter, errors

​Policies

​Spec types

​StreamSpec<T> accessors

​Not covered

​Validation

Overview

`StreamDefinitionContributor`

`StreamRegistration`

`StreamSpec.Builder<T>`

Target and behaviour

Handlers

Pipeline, filter, errors

Policies

Spec types

`StreamSpec<T>` accessors

Not covered

Validation