How to propagate context to downstream operators in Project Reactor?

user_2693851 0
Viewed 857

I need to propagate some context to downstream operators in Project Reactor, but it looks like they only allow to do so in a bottom-up fashion as described in the documentation.

Is there a way to propagate it from upstream operators to downstream? I can imaging a workaround with a wrapper like this, but I'd really like to avoid that:

private static class Contextualized<T> {
    private final Map<Object, Object> context;
    private final T content;

    public Contextualized(T content) {
        context = new HashMap<>();
        this.content = content;
    }

    public Contextualized(Map<Object, Object> context, T content) {
        this.context = context;
        this.content = content;
    }

    public Map<Object, Object> getContext() {
        return context;
    }

    public T getContent() {
        return content;
    }
}

public void someFunction() {
    Mono.just(new Contextualized<>(1))
            .map(it -> new Contextualized<>(Map.of("operation", "added 1"), it.getContent() + 1))
            .doOnNext(it -> System.out.println(it.getContext().get("operation").toString()))
            .subscribe();
    ...
}
project-reactor java
2 Answers

Your current solution is fine. You can go with that.

This could be another way. But might not be an elegant solution.

Flux.range(1, 5)
        .materialize() // convert the object to Signal<Object>
        .map(o -> {
            Context context = o.getContext().put("name", "somename");
            return o.isOnNext() ? Signal.next(o.get(), context) : Signal.complete(context);
        })
        ...
        ...
        .doOnNext(s -> System.out.println(s.getContext().get("name").toString())) // access ctx variables
        .dematerialize()  // convert the signal to regular object and lose the context here
user_28970080

You can try using subscriberContext

String key = "message";
Mono<String> r = Mono.just("Hello")
                .flatMap( s -> Mono.subscriberContext()
                                   .map( ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "World"));

StepVerifier.create(r)
            .expectNext("Hello World")
            .verifyComplete();
user_113526720
Related