@ChangeStream

The @ChangeStream annotation is the foundation of the FlowWarden programming model. It turns any class into a Spring-managed Change Stream handler that watches a MongoDB collection for real-time data changes.

Basic Usage

Annotate a class with @ChangeStream and add at least one event handler method (@OnChange, @OnInsert, @OnUpdate, @OnDelete, @OnReplace):

@ChangeStream(collection = "orders", documentType = Order.class)
public class OrderHandler {

    @OnChange
    void handleOrderChange(ChangeStreamContext<Order> ctx) {
        System.out.println(ctx.summary());
    }
}

@ChangeStream is meta-annotated with @Component — there is no need to add @Component or @Service separately. The class is automatically registered as a Spring bean.

Attributes Reference

Identification

Attribute	Type	Default	Description
`value`	`String`	`""`	Alias for `name`.
`name`	`String`	`""` (kebab-case of class name)	Unique stream name. Used in logs, checkpoints, and metrics.
`description`	`String`	`""`	Human-readable description. Surfaced by the FlowWarden Console when present.
`zone`	`String`	`""`	Grouping zone — used by the FlowWarden Console to organize related streams.

MongoDB Target

Attribute	Type	Default	Description
`collection`	`String`	`""`	Collection to watch. Required unless `documentType` has a Spring Data `@Document` annotation that provides the collection name.
`database`	`String`	`""`	Target database. Defaults to the Spring-configured database.
`documentType`	`Class<?>`	`Document.class`	Java type for deserialization. Also used to infer `collection` via `@Document` if not specified.

Document Options

Attribute	Type	Default	Description
`fullDocument`	`FullDocumentMode`	`DEFAULT`	Controls inclusion of the full document in change events. See values below.
`fullDocumentBeforeChange`	`FullDocumentBeforeChangeMode`	`OFF`	Controls pre-image inclusion (MongoDB 6.0+). See values below.

FullDocumentMode values

Value	Description
`DEFAULT`	Server default. INSERT and REPLACE include the document; UPDATE only includes the change delta.
`UPDATE_LOOKUP`	MongoDB looks up the current document for UPDATE events. Required when using `@Filter` with `@OnUpdate`.
`WHEN_AVAILABLE`	Include full document when available, `null` otherwise (MongoDB 6.0+).
`REQUIRED`	Require full document — error if not available (MongoDB 6.0+).

FullDocumentBeforeChangeMode values

Value	Description
`OFF`	Disabled (default). No pre-image is requested.
`WHEN_AVAILABLE`	Include the document before the change when available. Returns `Optional.empty()` if not enabled on the collection.
`REQUIRED`	Require the pre-image — throws an error if the collection does not have pre-images enabled.

Pre-images (fullDocumentBeforeChange) require MongoDB 6.0+ and that pre-images are explicitly enabled on the target collection:

db.runCommand({
  collMod: "orders",
  changeStreamPreAndPostImages: { enabled: true }
})

See the MongoDB documentation for details. Without this, WHEN_AVAILABLE returns empty and REQUIRED throws an error.

Deployment

Attribute	Type	Default	Description
`deploymentMode`	`DeploymentMode`	`ALL_INSTANCES`	Multi-instance coordination strategy. See the Deployment Modes guide for details.

DeploymentMode values

Value	Description
`ALL_INSTANCES`	Default. Every instance processes all events independently. No coordination.
`SINGLE_LEADER`	One leader instance processes events; others stand by. Uses MongoDB distributed locks (`_fw_locks`).

Behaviour

Attribute	Type	Default	Description
`enabled`	`boolean`	`true`	Enable or disable this stream.
`autoStart`	`boolean`	`true`	Automatically start the stream on application ready.
`mongoTemplateRef`	`String`	`""`	Name of the `MongoTemplate` or `ReactiveMongoTemplate` bean to use. Validated at startup.

Collection Resolution

If collection is not explicitly set, FlowWarden resolves it from documentType:

If documentType has a Spring Data @Document(collection = "...") annotation, that value is used.
Otherwise, the decapitalized simple class name is used (e.g. Order → order).
If neither works, startup fails with an explicit error.

// Explicit collection
@ChangeStream(collection = "orders")
public class OrderHandler { ... }

// Inferred from @Document annotation on Order class
@ChangeStream(documentType = Order.class)
public class OrderHandler { ... }

Name Resolution

The stream name is resolved in this order:

name() attribute if non-empty
value() attribute if non-empty
Kebab-case of the class simple name (e.g. OrderStreamHandler → order-stream-handler)

The stream name must be unique across all @ChangeStream classes in the application.

Always set a meaningful name — it appears in logs, metrics, and checkpoint keys. If omitted, FlowWarden generates one from the class name, which may break checkpoints if you rename the class.

Handler Methods

A @ChangeStream class must declare at least one handler method. FlowWarden supports two styles:

Generic: `@OnChange`

Catch-all for every operation that is not covered by a typed handler in the same class, including DROP and INVALIDATE. The annotation takes no attributes. Only one @OnChange method is allowed per class.

@ChangeStream(documentType = Order.class)
public class OrderHandler {

    @OnChange
    void handle(ChangeStreamContext<Order> ctx) {
        System.out.println(ctx.getOperationType() + " on " + ctx.getCollectionName());
    }
}

Typed: `@OnInsert`, `@OnUpdate`, `@OnDelete`, `@OnReplace`

Route events to specific methods by operation type. At most one method per typed annotation.

@ChangeStream(name = "typed-order-capture", documentType = Order.class)
public class TypedOrderHandler {

    @OnInsert
    void onInsert(Order doc, ChangeStreamContext<Order> ctx) {
        log.info("New order: {}", doc.getId());
    }

    @OnUpdate
    void onUpdate(ChangeStreamContext<Order> ctx) {
        log.info("Order updated: {}", ctx.getDocumentKey());
    }

    @OnDelete
    void onDelete(ChangeStreamContext<Order> ctx) {
        log.info("Order deleted: {}", ctx.getDocumentKey());
    }
}

Combining Both

When both typed handlers and @OnChange are present, typed handlers take priority. @OnChange acts as a fallback for operation types without a dedicated handler.

@ChangeStream(name = "typed-order-capture", documentType = Order.class)
public class OrderHandler {

    @OnInsert
    void onInsert(Order doc, ChangeStreamContext<Order> ctx) { ... }

    @OnUpdate
    void onUpdate(ChangeStreamContext<Order> ctx) { ... }

    @OnDelete
    void onDelete(ChangeStreamContext<Order> ctx) { ... }

    @OnChange
    void onFallback(ChangeStreamContext<Order> ctx) {
        // Catches REPLACE, DROP, INVALIDATE — anything without a typed handler above
    }
}

Supported Signatures

Typed handler methods (@OnInsert, @OnUpdate, @OnDelete, @OnReplace) support three signature styles:

Signature	Style	Description
`void handle(ChangeStreamContext<T> ctx)`	`CONTEXT_ONLY`	Access context; get the document from `ctx`.
`void handle(T doc)`	`DOCUMENT_ONLY`	Receive the deserialized document directly.
`void handle(T doc, ChangeStreamContext<T> ctx)`	`DOCUMENT_AND_CONTEXT`	Both the document and the context.

@OnChange only supports the CONTEXT_ONLY style: void handle(ChangeStreamContext ctx).

If you use DOCUMENT_ONLY or DOCUMENT_AND_CONTEXT signatures, you must set a concrete documentType on @ChangeStream (not Document.class). Otherwise, startup fails with a validation error.

Examples

Minimal

@ChangeStream(documentType = Order.class)
public class OrderStream {

    @OnChange
    void handleOrderChange(ChangeStreamContext<Order> ctx) {
        System.out.println(ctx.summary());
    }
}

@ChangeStream(documentType = Order.class)
public class OrderStream {

    @OnChange
    void handleOrderChange(ChangeStreamContext<Order> ctx) {
        System.out.println(ctx.summary());
    }
}

The handler class code is identical in both modes. The execution mode is determined globally by the flowwarden.default-mode property (IMPERATIVE or REACTIVE), which selects the appropriate stream manager at auto-configuration time.

With Explicit Name and Collection

@ChangeStream(name = "order-watcher", collection = "orders")
public class OrderHandler {

    @OnChange
    void handle(ChangeStreamContext<?> ctx) {
        System.out.printf("Event %s on %s%n",
            ctx.getOperationType(), ctx.getCollectionName());
    }
}

With Pre-Image (Before/After Comparison)

@ChangeStream(
    documentType = Order.class,
    fullDocument = FullDocumentMode.UPDATE_LOOKUP,
    fullDocumentBeforeChange = FullDocumentBeforeChangeMode.WHEN_AVAILABLE
)
public class OrderAuditHandler {

    @OnUpdate
    void onStatusChange(ChangeStreamContext<Order> ctx) {
        Optional<Order> before = ctx.getFullDocumentBeforeChange(Order.class);
        Optional<Order> after = ctx.getFullDocument(Order.class);

        if (before.isPresent() && after.isPresent()) {
            log.info("Order {} status: {} → {}",
                after.get().getId(),
                before.get().getStatus(),
                after.get().getStatus());
        }
    }
}

Disabled Stream

@ChangeStream(
    collection = "orders",
    enabled    = false   // Will not start automatically
)
public class OrderHandler { ... }

Custom MongoTemplate

When your application uses multiple MongoDB connections, specify which template to use:

@ChangeStream(
    collection       = "audit_events",
    mongoTemplateRef = "auditMongoTemplate"
)
public class AuditStreamHandler {

    @OnChange
    void handle(ChangeStreamContext<?> ctx) {
        // Uses the "auditMongoTemplate" bean
    }
}

Programmatic Start/Stop

Streams can be controlled at runtime via the FlowWardenStreamManager interface:

@Autowired
FlowWardenStreamManager streamManager;

// Stop a running stream
streamManager.stopStream("order-watcher");

// Restart it later
streamManager.startStream("order-watcher");

// Check status
boolean running = streamManager.isRunning("order-watcher");

Best Practices

One handler class per collection — keep stream handlers focused and cohesive.
Set documentType when you need typed access — otherwise documents arrive as org.bson.Document.
Use enabled = false to temporarily disable a stream without removing the code.
Set autoStart = false if you need to start the stream programmatically after some initialization logic.

MongoDB must be configured as a Replica Set for Change Streams to work. This applies to production and development environments. Testcontainers automatically provisions a single-node Replica Set for tests.

Event Handlers

@OnInsert, @OnUpdate, @OnDelete, @OnChange — handler method signatures in detail.

@Filter

Push server-side predicates to MongoDB’s aggregation pipeline.

@Checkpoint

Configure resume token persistence for crash-resilient streams.

Configuration

flowwarden.default-mode and other application properties.

Annotations

Registration API

Runtime

SPI

Basic Usage

Attributes Reference

Identification

MongoDB Target

Document Options

FullDocumentMode values

FullDocumentBeforeChangeMode values

Deployment

DeploymentMode values

Behaviour

Collection Resolution

Name Resolution

Handler Methods

Generic: `@OnChange`

Typed: `@OnInsert`, `@OnUpdate`, `@OnDelete`, `@OnReplace`

Combining Both

Supported Signatures

Examples

Minimal

With Explicit Name and Collection

With Pre-Image (Before/After Comparison)

Disabled Stream

Custom MongoTemplate

Programmatic Start/Stop

Best Practices

See Also

Event Handlers

@Filter

@Checkpoint

Configuration

​Basic Usage

​Attributes Reference

​Identification

​MongoDB Target

​Document Options

​FullDocumentMode values

​FullDocumentBeforeChangeMode values

​Deployment

​DeploymentMode values

​Behaviour

​Collection Resolution

​Name Resolution

​Handler Methods

​Generic: @OnChange

​Typed: @OnInsert, @OnUpdate, @OnDelete, @OnReplace

​Combining Both

​Supported Signatures

​Examples

​Minimal

​With Explicit Name and Collection

​With Pre-Image (Before/After Comparison)

​Disabled Stream

​Custom MongoTemplate

​Programmatic Start/Stop

​Best Practices

​See Also

Event Handlers

@Filter

@Checkpoint

Configuration

Basic Usage

Attributes Reference

Identification

MongoDB Target

Document Options

FullDocumentMode values

FullDocumentBeforeChangeMode values

Deployment

DeploymentMode values

Behaviour

Collection Resolution

Name Resolution

Handler Methods

Generic: `@OnChange`

Typed: `@OnInsert`, `@OnUpdate`, `@OnDelete`, `@OnReplace`

Combining Both

Supported Signatures

Examples

Minimal

With Explicit Name and Collection

With Pre-Image (Before/After Comparison)

Disabled Stream

Custom MongoTemplate

Programmatic Start/Stop

Best Practices

See Also