ExchangeAck

Utilities for acknowledging message exchanges.

Provides extension functions to ensure messages are acknowledged regardless of processing success or failure.

Functions

inline fun <T : MessageExchange<*, *>> Flux<T>.filterThenAck(crossinline predicate: (T) -> Boolean): Flux<T>

Filters the flux and acknowledges exchanges that don't match the predicate.

fun Flux<*>.finallyAck(exchange: MessageExchange<*, *>): Mono<Void>

Ensures the exchange is acknowledged after Flux completion, even on error.

fun Mono<*>.finallyAck(exchange: MessageExchange<*, *>): Mono<Void>

Ensures the exchange is acknowledged after Mono completion, even on error.