> ## 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.
# Filtering Events
> Reduce noise and CPU with @Pipeline (server-side) and @Filter (application-side)
FlowWarden provides two complementary filtering mechanisms. They answer a different question each:
| | `@Pipeline` | `@Filter` |
| ------------------- | ------------------------ | ------------------------------------------ |
| **Where it runs** | MongoDB server | Your JVM |
| **When it runs** | Once at stream startup | On every received event |
| **What it filters** | Events crossing the wire | Events dispatched to handlers |
| **Best for** | Reducing network traffic | Complex logic, Spring beans, service calls |
Use `@Pipeline` as your first line of defense — events that don't match never leave MongoDB. Use `@Filter` when the decision requires Java code that can't be expressed as an aggregation stage.
***
## @Pipeline — Server-side filtering
The `@Pipeline` method returns an aggregation pipeline that MongoDB applies directly in the oplog. Only matching events are sent to your application over the wire.
**The method is called once at stream startup** — the pipeline is static for the lifetime of the stream.
### Signatures
Three return types are supported:
Native MongoDB driver — most explicit, no Spring dependency:
```java theme={null}
@ChangeStream(documentType = Order.class)
public class OrderStream {
@Pipeline
List pipeline() {
return List.of(
Aggregates.match(Filters.in("operationType", "insert", "update"))
);
}
@OnInsert
void onInsert(Order doc, ChangeStreamContext ctx) { ... }
@OnUpdate
void onUpdate(ChangeStreamContext ctx) { ... }
}
```
Spring Data `Aggregation` — integrates with the Spring ecosystem:
```java theme={null}
@ChangeStream(documentType = Order.class, fullDocument = FullDocumentMode.UPDATE_LOOKUP)
public class OrderStream {
@Pipeline
Aggregation pipeline() {
return Aggregation.newAggregation(
Aggregation.match(
Criteria.where("operationType").in("insert", "update")
)
);
}
@OnInsert
void onInsert(Order doc, ChangeStreamContext ctx) { ... }
}
```
Spring Data operations list — useful to compose stages programmatically:
```java theme={null}
@Pipeline
List pipeline() {
return List.of(
Aggregation.match(Criteria.where("operationType").is("insert")),
Aggregation.match(Criteria.where("fullDocument.status").is("PAID"))
);
}
```
`List` and Spring Data `Aggregation` are the most common choices. Use `List` when you are already using the MongoDB driver directly; use `Aggregation` when you prefer to stay in the Spring Data world.
***
## @Filter — Application-side filtering
The `@Filter` method is a Java predicate evaluated **on every event received** by FlowWarden, before the event is dispatched to your handler. Events that don't pass the filter are silently dropped — their resume token is still tracked.
Because `@Filter` runs in Java, it can call Spring services, access request context, or apply any logic that cannot be expressed as a MongoDB aggregation stage.
### Signatures
Two styles are supported:
Takes a `ChangeStreamContext` and returns a `boolean` — the most readable style:
```java theme={null}
@ChangeStream(documentType = Order.class, fullDocument = FullDocumentMode.UPDATE_LOOKUP)
public class OrderStream {
@Filter
boolean filter(ChangeStreamContext ctx) {
return ctx.getFullDocument(Order.class)
.map(order -> order.getTotal().compareTo(new BigDecimal("100")) > 0)
.orElse(false);
}
@OnInsert
void onInsert(Order doc, ChangeStreamContext ctx) { ... }
@OnUpdate
void onUpdate(ChangeStreamContext ctx) { ... }
}
```
No-parameter method that returns a `Predicate` — useful to capture a Spring bean once:
```java theme={null}
@ChangeStream(documentType = Order.class, fullDocument = FullDocumentMode.UPDATE_LOOKUP)
public class OrderStream {
@Autowired
private TenantService tenantService;
@Filter
Predicate> filter() {
return ctx -> ((ChangeStreamContext) ctx)
.getFullDocument(Order.class)
.map(order -> tenantService.isActive(order.getTenantId()))
.orElse(false);
}
@OnInsert
void onInsert(Order doc, ChangeStreamContext ctx) { ... }
}
```
***
## Combining both — the funnel approach
`@Pipeline` and `@Filter` can coexist on the same `@ChangeStream`. MongoDB pre-filters server-side, then Java refines application-side. This is the most efficient pattern for high-throughput streams with complex filtering needs.
```mermaid theme={null}
flowchart LR
A[("Full oplog")] -->|"all events"| B["@Pipeline
MongoDB server-side
reduces network traffic"]
B -->|"matched events only"| C["@Filter
Java application-side
complex logic"]
C -->|"passing events"| D["Handler"]
C -->|"filtered events"| X(["dropped"])
```
**Example** — pre-filter by operation type server-side, then check business logic in Java:
```java theme={null}
@ChangeStream(documentType = Order.class, fullDocument = FullDocumentMode.UPDATE_LOOKUP)
public class OrderStream {
@Autowired
private TenantService tenantService;
/** Server-side: only INSERT and UPDATE events leave MongoDB */
@Pipeline
Aggregation pipeline() {
return Aggregation.newAggregation(
Aggregation.match(Criteria.where("operationType").in("insert", "update"))
);
}
/** Application-side: only orders from active tenants reach the handler */
@Filter
Predicate> filter() {
return ctx -> ((ChangeStreamContext) ctx)
.getFullDocument(Order.class)
.map(order -> tenantService.isActive(order.getTenantId()))
.orElse(false);
}
@OnInsert
void onInsert(Order doc, ChangeStreamContext ctx) { ... }
@OnUpdate
void onUpdate(ChangeStreamContext ctx) { ... }
}
```
***
## Constraints and pitfalls
**`@Filter` requires `fullDocument` for UPDATE events.**
By default (`fullDocument = DEFAULT`), MongoDB does not include the full document in UPDATE events. If your `@Filter` accesses `getFullDocument()` on an UPDATE event, it will receive `Optional.empty()` and likely drop the event silently.
Set `fullDocument = FullDocumentMode.UPDATE_LOOKUP` on `@ChangeStream` when using `@Filter` with `@OnUpdate`:
```java theme={null}
@ChangeStream(documentType = Order.class, fullDocument = FullDocumentMode.UPDATE_LOOKUP)
```
**`@Filter` is incompatible with typed handlers for no-fullDocument operations (e.g. `@OnDelete`).**
DELETE, DROP, and INVALIDATE events never carry a `fullDocument`. Combining `@Filter` with a typed handler dedicated to one of these operations is rejected at startup with a `BeanCreationException`.
Combining `@Filter` with `@OnChange` is allowed — but DROP and INVALIDATE will still reach the predicate, which must guard against `ctx.getFullDocument().isEmpty()`. If you need both filtering and DELETE handling without that guard, use a server-side `@Pipeline` to filter, or move the filtering logic inside the handler method itself.
Events rejected by `@Filter` are silently dropped — they never reach any handler method. Their resume token is still tracked internally, so checkpointing is not affected. Filtered events are not sent to the Dead Letter Queue.
If you need to log or count filtered events, use the `boolean filter(ChangeStreamContext ctx)` style and add your own logging before returning `false`.
The most common first stage in a `@Pipeline` is filtering by `operationType`:
```java theme={null}
Aggregates.match(Filters.in("operationType", "insert", "update"))
```
The app-side equivalent is to declare typed handlers (e.g. `@OnInsert` + `@OnUpdate`, stacked on one method if they share logic) — non-matching ops are silently skipped. Filtering server-side in `@Pipeline` is more efficient when most events would otherwise be discarded, and it lets you combine the operation filter with additional field conditions in the same `$match`.
***
## See Also
All supported return types and validation rules
Signature styles and startup validation
Where filtering fits in the event processing pipeline
How filtered events interact with resume tokens