How to enrich Reactor's Context with data from Mono?

user_7781169 0
Viewed 1403

There is a possibility to enrich Reactor's sequence context with some data at subscription time as follows:

Mono.just("Hello")
    .flatMap(s -> Mono.deferContextual(ctx -> Mono.just(s + " " + ctx.get(key)))) 
    .contextWrite(ctx -> ctx.put(key, "World"));

However, this "World" here is sort of static data which is know at subscription time.

I'd like to resolve a value from the initial Mono itself and put it into the context, so that downstream operators can access it via context API.

How can I properly do that?

The use case I am working on is following:

  • I receive an SQS message from AWS via the async SDK which is resolved to a Reactor's Mono
  • A message has a correlation_id inside which I want to extract and put it into Reactors context to make it visible to all downstream operators.
project-reactor java
1 Answers

For now, the only option to enrich Reactor's context with data from the initial Mono I have found
is to wrap all processing into a flatMap with Mono.deferContextual()...contextWrite() inside as following:

Mono.just("test")
        .flatMap(s ->
                Mono.deferContextual(ctx -> {
                    System.out.println("key1=" + ctx.getOrEmpty("key")); // Context in place
                    return Mono.just(s);
                }).contextWrite(ctx -> ctx.put("key", s)))
        .flatMap(s -> Mono.deferContextual(ctx -> {
            System.out.println("key2=" + ctx.getOrEmpty("key")); // No Context
            return Mono.just(s);
        }));

I hope someone suggests a better solution.

user_77811690
Related