Reactor-core: Subscription Context

[sskrla]

I have a use case where by I need context to follow a subscription across schedulers.

Specifically, each request that comes to a service has a X-Request-ID used to correlate logs across services. As data flows down a subscription it may call out to other services, usually using some sort of a scheduler. Traditionally, we would use a thread local set by the requesting http handler thread to look up any kind of context that needed to be passed as part of the call. This could include credentials in addition to the request Id in question.

I realize that context data is not very "functional" and can result in all manner of bugs, but I also do not believe things like authentication and or debug information should be forced in some sort of wrapper that dilutes the functions / API with signatures like Flux<RequestContext<T>>

What I had in mind was something to the effect of:

void takeAction(@Context ServletRequest request, ActionRequest action) {
    Flux.from(action)
        .flatMap(a -> externalService.doThatThing(a.thing()))
        .subscribeOn(httpClients)
        .subscribe(ImmutableMap.of("request", request))
}

class ExternalService {
    Mono<Response> doThatThing(String thing) {
        String txId = Subscription.getMetadata("request", ServletRequest).getHeader("X-Request-ID");
        ....
    }
}

Alternatively if there were some way of observing the subscription beyond .onSubscribe I could manage my own context:

void takeAction(@Context ServletRequest request, ActionRequest action) {
    Flux.from(action)
        .compose(SubscriptionContext::apply)
        .flatMap((subscription, action) -> // New bifunction alternative
            externalService.doThatThing(
                SubscriptionContext.get(subscription).getRequest(), 
                a.thing())))
        .subscribeOn(httpClients)
        .subscribe(ImmutableMap.of("request", request))
}

Another alternative exposing the subscription:

void takeAction(@Context ServletRequest request, ActionRequest action) {
    Flux.from(action)
        .compose(SubscriptionContext::apply)
        .withContext((sub, flux) -> flux.flatMap(action->
            externalService.doThatThing(
                SubscriptionContext.get(subscription).getRequest(), 
                a.thing())))
        .subscribeOn(httpClients)
        .subscribe(ImmutableMap.of("request", request))
}

[smaldini]

After 3.0 RC1 this could be looked into since its not breaking - any feedback @akarnokd ?

[akarnokd]

I don't understand the issue. I use composite objects or lambda capture of the extra data.

[sskrla]

Lambda captures are not an option here unless I misunderstand you. In most cases, this is shared code in a separate class with no lexical scope.

As for composing, it causes a significant amount of refactoring work and, I feel, leaks implementation details that do not pertain directly to the API.

I realize that a purely functional implementation should have everything you need for a function to accomplish its work in a side effect free way. However, I think it is a rather cumbersome requirement, and contexts are immensely useful for simplifying code and keeping diagnostic features out of the way.

While first-class support for the concept would be fantastic, all I am asking for is an extension mechanic where I can decorate the subscription with information and get at the information.

[smaldini]

Support added in #447
Extending this with specific operators (now provided with contextMap()) see https://github.com/reactor/reactor-core/blob/master/reactor-core/src/test/java/reactor/core/publisher/BaseSubscriberTest.java#L136