Operator intercept()
Starting with version 5.3, the intercept() operator allows to register one or more ChannelInterceptor instances at the current MessageChannel in the flow.
This is an alternative to creating an explicit MessageChannel via the MessageChannels API.
The following example uses a MessageSelectingInterceptor to reject certain messages with an exception:
.transform(...)
.intercept(new MessageSelectingInterceptor(m -> m.getPayload().isValid()))
.handle(...)