Pass a context from one flux/mono to another

Question

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 1

You want to do a couple of things:

1.) Publish a subscriber context

mono.subscriberContext({ Context context ->
    context.put("key", "value")
})

2.) Subscribe/access a subscriber context

mono.flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        context.get("key")
        context.get("keyOrMapOrStateObject").put("someKey", "someData")
        return r
    })
})

3.) Potentially pass data from one event to a downstream event

mono.flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        def someData = context.get("keyOrMapOrStateObject").get("someKey")
        return r
    })
})

All together it'll look something like: (this is groovy syntax)

mono.flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        context.get("key")
        context.get("keyOrMapOrStateObject").put("someKey", "someData")
        return r
    })
}).flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        def someData = context.get("keyOrMapOrStateObject").get("someKey")
        return r
    })
}).subscriberContext({ Context context ->
    context.put("key", "value")
    context.put("keyOrMapOrStateObject", new HashMap())
})

This is a rough outline - not ready 'as is' but it should help you understand the pattern.

Good luck!

Answer 2

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.