Spring webflux reactive Mono::subscribe

Question

Problem statement:

I am trying to save session in redis before the request is forwarded to downstream system.

 return new OrderedGatewayFilter((exchange, chain) -> exchange.getSession().doOnNext(session -> {
     // some code
}
}).map(WebSession::save).doOnNext(Mono::subscribe).then(chain.filter(exchange)), SAVE_SESSION_FILTER);

However Mono::subscribe is asynchornous is nature, and there is no guarantee that session will be stored in redis first andthen forwarded to downstream. We are facing this issue while performing performance testing with concurrent request.

Is there any way to make it synchronous.

Answer 1

You are probably looking for flatMap operator which makes the subscribe unnecessary and will not progress the flow until it is finished:

return new OrderedGatewayFilter((exchange, chain) ->
    exchange.getSession()
        .doOnNext(session -> {/* some code*/})
        .flatMap(WebSession::save)
        .then(chain.filter(exchange)), SAVE_SESSION_FILTER);