> ## Documentation Index > Fetch the complete documentation index at: https://docs.flowwarden.io/llms.txt > Use this file to discover all available pages before exploring further. # Programmatic Registration > Declare Change Streams without annotations — from YAML, a database, or a feature-flag service — with StreamDefinitionContributor and StreamSpec `@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 ```java theme={null} @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: ```java theme={null} @ChangeStream(name = "order-stream", collection = "orders", documentType = Order.class) @Checkpoint public class OrderStream { @OnInsert void onInsert(Order order, ChangeStreamContext 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: ```yaml application.yml theme={null} streams: catalog: - name: confirmed-orders collection: orders keep-status: CONFIRMED dlq: true - name: shipments collection: shipments ``` ```java StreamCatalogContributor.java theme={null} @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 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](https://github.com/flowwarden-io/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` (context only) or `DocumentHandler` (document + context) | | `@OnChange` | `.onChange(..)` | Catch-all, `ContextHandler` | | Reactive handlers | `.onInsertReactive(..)`, … `.onChangeReactive(..)` | Same shapes, returning `Mono` | | `@Pipeline` | `.pipeline(Supplier>)` | Evaluated once at stream start; one per stream | | `@Filter` | `.filter(Predicate>)` | One per stream; rejected with a typed handler on DELETE/DROP/INVALIDATE, like the annotation | | `@OnError` | `.onError(ErrorHandler, Class...)` | 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](/reference/stream-spec) for every signature. *** ## Pipeline, filter and error handling The three annotation capabilities that need code, not just attributes, take functional interfaces: ```java theme={null} 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`](/reference/on-error). 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.