Pass a context from one flux/mono to another

java spring-webflux project-reactor
Gie Spaepen picture Gie Spaepen · Jul 5, 2018 · Viewed 7.7k times · Source

Interesting things happen inside the webflux package. However, my journey in the source didn't solve the following question.

Let's say I have the following mono (or flux):

Mono hello = Mono.empty()
            .subscriberContext(ctx -> ctx.put("message", "hello"));

I use similar construct within a webfilter to enrich the pipeline with tenant and user data. Then in a controller a construct like this is used:

Mono world = Mono.subscriberContext()
            .map(ctx -> (String)ctx.get("message"));

The context of the hello mono is filled in the world mono. I tried to figure out how this is done, also for unit testing purposes.

In the end it remains a riddle. I tried to do this with the regular methods available on both mono/flux objects but I don't succeed in making the hello context available to the world mono. How can you fuse fluxes and monos and passing the context along the way to the upstream operators?

Answer

Simon Baslé picture Simon Baslé · Jul 5, 2018

WebFlux takes your world Mono and builds a reactive chain on top of it, with an HTTP request within reactor-netty as the ultimate source. The WebFilter are part of the chain construction, so they can enrich the Context of the whole chain.

IIRC Mono.subscriberContext() will be used within a flatMap, which makes the main sequence Context available to its inners, so it can see the Context of hello.

Related questions

How to correctly read Flux<DataBuffer> and convert it to a single inputStream

I'm using WebClient and custom BodyExtractorclass for my spring-boot application WebClient webLCient = WebClient.create(); webClient.get() .uri(url, params) .accept(MediaType.APPLICATION.XML) .exchange() .flatMap(response -> { return response.body(new BodyExtractor()); }) BodyExtractor.java @Override public Mono<T&…

Read More
block()/blockFirst()/blockLast() are blocking error when calling bodyToMono AFTER exchange()

I am trying to use Webflux to stream a generated file to another location, however, if the generation of the file ran into an error, the api returns success, but with a DTO detailing the errors while generating the file …

Read More
Spring WebFlux, how can I debug my WebClient POST exchange?

I am having trouble understanding what I've done wrong in constructing my WebClient request. I would like to understand what the actual HTTP request looks like. (e.g., dumping the raw request to console) POST /rest/json/send HTTP/1.1 Host: …

Read More