Programmatic Registration

@ChangeStream is the natural way to declare a stream when its shape is known at compile time. It stops being enough when the stream catalog lives outside the JVM: a YAML file an operator edits, a table in a configuration database, a feature-flag service that turns streams on per tenant. Since 1.0.0-rc.6, the io.flowwarden.stream.registration package covers that case. A StreamDefinitionContributor bean receives a StreamRegistration at bootstrap and describes streams with a builder, StreamSpec. No annotated class is involved, and the contributed streams go through the same validation, defaults, and runtime as annotated ones.

Registration is bootstrap-only. Contributors run once, after every singleton bean has been created (annotated @ChangeStream classes included) and before the stream managers read the catalog. A contributed stream is fixed for the lifetime of the application context — there is no hot registration on a running instance.

Minimal example

@Component
class OrderStreamContributor implements StreamDefinitionContributor {

    private final OrderService orderService;

    OrderStreamContributor(OrderService orderService) {
        this.orderService = orderService;
    }

    @Override
    public void contribute(StreamRegistration registration) {
        registration.stream("order-stream", Order.class)
                .collection("orders")
                .checkpoint(CheckpointSpec.defaults())
                .onInsert((order, ctx) -> orderService.onNewOrder(order));
    }
}

That is the equivalent of:

@ChangeStream(name = "order-stream", collection = "orders", documentType = Order.class)
@Checkpoint
public class OrderStream {

    @OnInsert
    void onInsert(Order order, ChangeStreamContext<Order> ctx) {
        orderService.onNewOrder(order);
    }
}

The contributor is an ordinary Spring bean: inject whatever the handlers need and capture it in the lambdas.

Driving the catalog from configuration

The point of the API is that the number and shape of streams can come from data. A @ConfigurationProperties class bound to application.yml is the simplest source:

streams:
  catalog:
    - name: confirmed-orders
      collection: orders
      keep-status: CONFIRMED
      dlq: true
    - name: shipments
      collection: shipments

@Component
class StreamCatalogContributor implements StreamDefinitionContributor {

    private final StreamCatalogProperties catalog;
    private final OrderService orderService;

    // constructor omitted

    @Override
    public void contribute(StreamRegistration registration) {
        for (StreamCatalogProperties.Entry entry : catalog.getCatalog()) {
            StreamSpec.Builder<Order> stream = registration
                    .stream(entry.getName(), Order.class)
                    .collection(entry.getCollection())
                    .checkpoint(CheckpointSpec.defaults())
                    .onInsert((order, ctx) -> orderService.onNewOrder(entry.getName(), order));

            if (entry.getKeepStatus() != null) {
                String keep = entry.getKeepStatus();
                stream.filter(ctx -> ctx.getFullDocument(Order.class)
                        .map(o -> keep.equals(o.getStatus()))
                        .orElse(false));
            }
            if (entry.isDlq()) {
                stream.deadLetterQueue(DeadLetterQueueSpec.defaults());
            }
        }
    }
}

Change the YAML, restart: the stream set follows. Nothing is recompiled. The full runnable version of this pattern is sample 12-registration in flowwarden-examples, in both imperative and reactive flavours.

Annotation ↔ builder map

Every builder call maps 1:1 to an annotation, with the same defaults and the same fail-fast rules.

Annotation	`StreamSpec.Builder`	Notes
`@ChangeStream(collection, database, mongoTemplateRef, …)`	`.collection(..)`, `.database(..)`, `.mongoTemplateRef(..)`, `.enabled(..)`, `.autoStart(..)`, `.fullDocument(..)`, `.fullDocumentBeforeChange(..)`, `.deploymentMode(..)`	`name` and `documentType` are the two builder arguments
`@OnInsert` / `@OnUpdate` / `@OnDelete` / `@OnReplace`	`.onInsert(..)`, `.onUpdate(..)`, `.onDelete(..)`, `.onReplace(..)`	Two shapes each: `ContextHandler<T>` (context only) or `DocumentHandler<T>` (document + context)
`@OnChange`	`.onChange(..)`	Catch-all, `ContextHandler<T>`
Reactive handlers	`.onInsertReactive(..)`, … `.onChangeReactive(..)`	Same shapes, returning `Mono<Void>`
`@Pipeline`	`.pipeline(Supplier<List<Bson>>)`	Evaluated once at stream start; one per stream
`@Filter`	`.filter(Predicate<ChangeStreamContext<T>>)`	One per stream; rejected with a typed handler on DELETE/DROP/INVALIDATE, like the annotation
`@OnError`	`.onError(ErrorHandler, Class<? extends Throwable>...)`	Repeatable; no types = catch-all (at most one); a type may be claimed by one handler only
`@Checkpoint`	`.checkpoint(CheckpointSpec)`	`CheckpointSpec.defaults()` or `CheckpointSpec.builder()`
`@RetryPolicy`	`.retryPolicy(RetryPolicySpec)`	idem
`@DeadLetterQueue`	`.deadLetterQueue(DeadLetterQueueSpec)`	idem
`@MongoDlqOptions`	`.mongoDlqOptions(MongoDlqOptionsSpec)`	idem
`zone`	—	Not covered yet. A stream needing a zone stays annotated.

See the StreamSpec reference for every signature.

Pipeline, filter and error handling

The three annotation capabilities that need code, not just attributes, take functional interfaces:

registration.stream("high-value-orders", Order.class)
        .collection("orders")
        // @Pipeline — server-side, once at start
        .pipeline(() -> List.of(
                Aggregates.match(Filters.eq("operationType", "insert"))))
        // @Filter — client-side, every event
        .filter(ctx -> ctx.getFullDocument(Order.class)
                .map(o -> o.getTotal() > 1_000)
                .orElse(false))
        // @OnError(IllegalStateException.class)
        .onError((ex, ctx) -> ErrorAction.SKIP, IllegalStateException.class)
        // catch-all @OnError
        .onError((ex, ctx) -> ErrorAction.RETHROW)
        .retryPolicy(RetryPolicySpec.builder().maxAttempts(5).build())
        .deadLetterQueue(DeadLetterQueueSpec.defaults())
        .onInsert((order, ctx) -> billing.charge(order));

ErrorHandler is a public functional interface in io.flowwarden.stream.core: ErrorAction handle(Throwable ex, ChangeStreamContext<?> ctx). Resolution order between scoped and catch-all handlers, and the meaning of each ErrorAction, are the same as for @OnError.

A functional handler that throws — in either mode, Error included — is routed through @OnError / retry / DLQ resolution exactly like an annotated method. An ErrorHandler that itself throws falls back to RETHROW.

Validation and failure modes

A contributed stream is validated at bootstrap with the rules shared with the annotation path:

checkpoint, retry, and DLQ bounds;
collection resolution (collection, or a documentType annotated with @Document; the raw Document.class without a collection fails);
mongoTemplateRef must name a bean that is a MongoTemplate / ReactiveMongoTemplate;
handler mode must match the execution mode (imperative handlers on an imperative stream, reactive on reactive);
a filter cannot be combined with a typed handler on an operation without a fullDocument;
duplicate onError types or catch-alls are rejected.

A duplicate stream name — annotated vs contributed, or two contributors — fails application startup. So does a builder misuse: a second pipeline, a second filter, or a second handler for the same operation throws IllegalStateException from the builder itself.

When to stay on annotations

The stream needs a zone — not on StreamSpec yet.
The stream is one fixed unit of code with no external configuration: annotations stay shorter and are visible in the class.
You want the stream to appear in code search by its handler methods.

Both styles coexist in the same application and share the same catalog, actuator endpoints and metrics.

Getting Started

FAQ

Guides

Programmatic Registration

Minimal example

Driving the catalog from configuration

Annotation ↔ builder map

Pipeline, filter and error handling

Validation and failure modes

When to stay on annotations

​Minimal example

​Driving the catalog from configuration

​Annotation ↔ builder map

​Pipeline, filter and error handling

​Validation and failure modes

​When to stay on annotations

Minimal example

Driving the catalog from configuration

Annotation ↔ builder map

Pipeline, filter and error handling

Validation and failure modes

When to stay on annotations