How it Works

FlowWarden wraps MongoDB Change Streams into a declarative, annotation-driven programming model managed by Spring Boot auto-configuration. This page explains what happens under the hood — from application startup to handler invocation.

The Big Picture

Phase 1 — Startup: Bean Discovery

When the application context boots, Spring initializes all beans. ChangeStreamBeanPostProcessor is a BeanPostProcessor registered by the auto-configuration — it intercepts every bean after initialization and inspects its class for the @ChangeStream annotation. For each discovered @ChangeStream class, the processor:

Resolves the stream name and target collection — from name, value, collection, or inferred from documentType via the Spring Data @Document annotation.
Discovers handler methods — scans declared methods for @OnChange, @OnInsert, @OnUpdate, @OnDelete, and @OnReplace, validates their signatures and return types.
Discovers the pipeline — finds the optional @Pipeline method and inspects its return type (List<Bson>, List<AggregationOperation>, or Aggregation).
Discovers the filter — finds the optional @Filter method and checks it returns Predicate<ChangeStreamContext<?>> or boolean filter(ChangeStreamContext<?> ctx).
Validates the configuration — catches misconfigurations early (e.g., @Filter combined with @OnDelete, missing handlers, wrong return types for the configured mode).
Registers a ChangeStreamDefinition in the StreamRegistry.

All validation happens at startup, not at runtime. Misconfigured streams fail fast with a clear BeanCreationException before the application finishes booting.

Phase 2 — Startup: Stream Launch

Once the application context is fully ready (ApplicationReadyEvent), the StreamManager iterates the StreamRegistry and starts every stream where autoStart = true and enabled = true.

// ImperativeStreamManager — uses Spring Data's MessageListenerContainer
MessageListenerContainer container = new DefaultMessageListenerContainer(mongoTemplate);
container.start();

ChangeStreamRequest<Document> request = ChangeStreamRequest.builder(listener)
    .collection(def.collection())
    .pipeline(pipeline)          // @Pipeline stages, evaluated once here
    .resumeToken(lastToken)      // null on first start; loaded from _fw_checkpoints on restart
    .build();

container.register(request, Document.class);

// ReactiveStreamManager — uses Project Reactor + ReactiveMongoTemplate
reactiveMongoTemplate
    .changeStream(def.collection(), options, Document.class)
    .doOnNext(event -> dispatch(def, event))
    .subscribe();

If a checkpoint was previously saved in _fw_checkpoints, the resume token is loaded and passed to MongoDB so the stream resumes exactly where it left off — no events are missed or replayed.

Phase 3 — Runtime: Event Processing Pipeline

Every Change Stream event flows through the following stages before reaching your handler:

@Pipeline vs @Filter

@Pipeline — Server-side
@Filter — Application-side
Combined — Funnel approach

Pushed to MongoDB at stream subscription time. MongoDB evaluates the aggregation pipeline in the oplog and only sends matching events over the wire. This is the most efficient approach — it reduces network traffic and application load.

@Pipeline
List<Bson> pipeline() {
    return List.of(
        Aggregates.match(Filters.in("operationType", "insert", "update")),
        Aggregates.match(Filters.eq("fullDocument.status", "PAID"))
    );
}

Evaluated by FlowWarden on every event received, after @Pipeline. Useful for logic that requires Spring bean injection or cannot be expressed as a MongoDB aggregation.

@Filter
boolean filter(ChangeStreamContext<Order> ctx) {
    return tenantService.isActive(ctx.getFullDocument(Order.class).getTenantId());
}

The two approaches are complementary. Use @Pipeline to reduce network traffic, then @Filter for precise application-level control.

Change Stream pipelines scan the oplog, not a collection. Unlike regular MongoDB queries, aggregation stages in a Change Stream pipeline cannot use indexes — MongoDB performs a COLLSCAN on the oplog to evaluate your $match filters. This is expected behavior, not a performance issue. Keep your pipeline stages lightweight ($match, $project) and avoid heavy operations like $lookup or $group. For complex filtering logic, use @Filter on the application side instead.

Dual Execution Mode

FlowWarden supports two execution modes, configured via flowwarden.default-mode.

Mode	Infrastructure	Handler return type	Use with
`IMPERATIVE`	`MongoTemplate` + `MessageListenerContainer`	`void`	Spring MVC, blocking I/O
`REACTIVE`	`ReactiveMongoTemplate` + Project Reactor	`Mono<Void>`	Spring WebFlux, non-blocking I/O

The mode is validated at startup: a void handler in REACTIVE mode or a Mono<Void> handler in IMPERATIVE mode causes a BeanCreationException.

@ChangeStream(collection = "orders")
public class OrderStream {

    @OnInsert
    void onInsert(Order order, ChangeStreamContext<Order> ctx) {
        // blocking code — fine in IMPERATIVE mode
        repository.save(order);
    }
}

@ChangeStream(collection = "orders")
public class OrderStream {

    @OnInsert
    Mono<Void> onInsert(Order order, ChangeStreamContext<Order> ctx) {
        // non-blocking — required in REACTIVE mode
        return repository.save(order).then();
    }
}

Internal MongoDB Collections

FlowWarden uses three internal collections in your database. You do not need to create them — they are managed automatically.

Collection	Purpose
`_fw_checkpoints`	Stores the dual-token checkpoint per stream (`lastSeenToken` + `lastProcessedToken`). Loaded on startup and consumed by the resume cascade.
`_fw_dlq`	Dead Letter Queue. Failed events that exhausted all retry attempts are written here.
`_fw_locks`	Distributed locks for `SINGLE_LEADER` deployment mode. Used for leader election and heartbeat coordination between instances.

All internal collections are prefixed with _fw_ to avoid collisions with your application data. They are created with appropriate indexes on first use.

Handler Signatures

FlowWarden resolves the correct invocation strategy at startup based on the method signature.

Typed handler signatures (@OnInsert, @OnUpdate, @OnDelete, @OnReplace)

Signature	Description
`void handle(ChangeStreamContext<T> ctx)`	Full event context — operation type, document key, update description, resume token
`void handle(T doc)`	Document directly — requires `documentType` on `@ChangeStream`
`void handle(T doc, ChangeStreamContext<T> ctx)`	Both document and context
`Mono<Void> handle(ChangeStreamContext<T> ctx)`	Reactive variant (REACTIVE mode only)

@OnChange — generic handler

@OnChange is a pure catch-all: it handles every operation type not covered by a typed handler in the same class — including DROP and INVALIDATE. It always receives a ChangeStreamContext<?> and takes no attributes.

@OnChange
void handle(ChangeStreamContext<?> ctx) {
    // handles INSERT, UPDATE, DELETE, REPLACE, DROP, INVALIDATE, ...
}

To restrict a handler to a specific subset of operations, declare typed handlers instead (@OnInsert, @OnUpdate, etc.) — stacked on a single method if they share logic.

@ChangeStream

Full annotation reference with all attributes

Checkpoint & Resume

How resume tokens are stored and replayed on restart

Filtering Events

When to use @Pipeline vs @Filter

Imperative vs Reactive

Choosing the right execution mode

Getting Started

FAQ

Guides

The Big Picture

Phase 1 — Startup: Bean Discovery

Phase 2 — Startup: Stream Launch

Phase 3 — Runtime: Event Processing Pipeline

@Pipeline vs @Filter

Dual Execution Mode

Internal MongoDB Collections

Handler Signatures

See Also

@ChangeStream

Checkpoint & Resume

Filtering Events

Imperative vs Reactive

​The Big Picture

​Phase 1 — Startup: Bean Discovery

​Phase 2 — Startup: Stream Launch

​Phase 3 — Runtime: Event Processing Pipeline

​@Pipeline vs @Filter

​Dual Execution Mode

​Internal MongoDB Collections

​Handler Signatures

​See Also

@ChangeStream

Checkpoint & Resume

Filtering Events

Imperative vs Reactive

The Big Picture

Phase 1 — Startup: Bean Discovery

Phase 2 — Startup: Stream Launch

Phase 3 — Runtime: Event Processing Pipeline

@Pipeline vs @Filter

Dual Execution Mode

Internal MongoDB Collections

Handler Signatures

See Also