> ## 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.

# StreamSpec

> Programmatic stream registration — StreamDefinitionContributor, StreamRegistration, StreamSpec.Builder and the spec types

## 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](/guides/programmatic-registration).

<Note>
  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.
</Note>

## `StreamDefinitionContributor`

```java theme={null}
@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`):

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

Resolution and `ErrorAction` semantics are those of [`@OnError`](/reference/on-error). 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.

<AccordionGroup>
  <Accordion title="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`](/reference/checkpoint) for the semantics and bounds.
  </Accordion>

  <Accordion title="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`](/reference/retry-policy).
  </Accordion>

  <Accordion title="DeadLetterQueueSpec">
    | Builder method                     | Annotation attribute                       |
    | ---------------------------------- | ------------------------------------------ |
    | `enabled(boolean)`                 | `@DeadLetterQueue.enabled`                 |
    | `retentionDays(int)`               | `@DeadLetterQueue.retentionDays`           |
    | `includeOriginalDocument(boolean)` | `@DeadLetterQueue.includeOriginalDocument` |
    | `includeStackTrace(boolean)`       | `@DeadLetterQueue.includeStackTrace`       |

    See [`@DeadLetterQueue`](/reference/dead-letter-queue).
  </Accordion>

  <Accordion title="MongoDlqOptionsSpec">
    | Builder method       | Annotation attribute          |
    | -------------------- | ----------------------------- |
    | `collection(String)` | `@MongoDlqOptions.collection` |
  </Accordion>
</AccordionGroup>

## `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.