Retaining context from in reactive streams

Question

Can someone help me understand how context is passed on in reactive streams. See below code for example:

Flux<Integer> expectedFluxWithContext = Flux.just(1, 2, 3, 4)
        .flatMap(item -> Mono.just(item).contextWrite(Context.of("traceId", item)))
        .doOnEach(signal -> System.out.println(signal.getContextView()));

When I run the above code I am getting the correct data as I would expect but in doOnEach operator the context is empty.

Can anyone help me to understand how context is shared in streams and what changes I can make to make this work.

Answer 1

context is only available "upwards" in the pipeline as it is attached at subscribe time, and that does not know about anything downstream. It is not meant to pass things that are acquired during the pipeline execution, if you want to "pass" that, you might want to use Tuples.

For more info I suggest reading the appropriate parts of the Reactor reference guide: https://projectreactor.io/docs/core/release/reference/#context