Reactor-core: subscriberContext is useless in current state (subjective opinion)

[apischan]

Expected behavior

It is able to put value to context just after getting it (in top of reactive stream) and get it later (in the bottom of reactive stream).

Actual behavior

onError(java.util.NoSuchElementException: Context is empty) instead of expected behavior.

Steps to reproduce

@Test
public void monoContext() {
    Flux<String> stringFlux = getValueFromDatabaseThatHaveReactiveInterface()
            .flatMap(hello -> Mono.subscriberContext().map(ctx -> ctx.put("someTopic", hello)))
            .thenMany(Flux.just("Hello", "world"))
            .concatMap(str -> Mono.subscriberContext().map(ctx -> str + ctx.get("someTopic")));

    StepVerifier.create(stringFlux)
            .expectNext("Hello hello")
            .expectNext("world hello")
            .verifyComplete();
}

private Mono<String> getValueFromDatabaseThatHaveReactiveInterface() {
    return Mono.just("hello");
}

Reactor Core version

3.1.6

JVM version (e.g. java -version)

java version "1.8.0_112"
Java(TM) SE Runtime Environment (build 1.8.0_112-b15)
Java HotSpot(TM) 64-Bit Server VM (build 25.112-b15, mixed mode)

So there is a big chance that I do not understand how to handle my case but with current state of things I don't see the solution.
I've read the documentation and as I understand that my case is impossible to solve with subscriberContext but maybe I am wrong, please show me how.

I want to comment a bit my goal. Suppose we have to get some value from database, do something with it and then to use it later in chain. So obvious solution would be to place it to some context and then (when it will be needed) get it from there and use for some purpose. So now I see the only one solution is to wrap value to some wrapper (e.g. Tuple2) and pass it through all the chain (which could be not so small). To be more specific - the last call in my chain from test could be:
.concatMap(tuple -> Mono.just(tuple.getLeft() + tuple.getRight()))

Now I see that subscriberContext works well in very simple synchronous cases where value that we desire to put into context is not wrapped to Mono or Flux. Otherwise we in trouble.

[OlegDokuka]

Subscriber context was designed as the immutable map which is created and transferred from the downstream to the upstream at the subscriptions time. The central idea of the Context passing contextual parameters, created, for example, at the HTTP request handling. In such cases, we may easily pass request-related information throughout the whole stream without holding it in the additional Tuple object or so.

On the one hand, such interaction model simplifies the overall design and put own limitations as well. On the other hand, to resolve mentioned problem you may do following:

@Test
public void monoContext() {
    Flux<String> stringFlux = Flux.just("Hello ", "world ")
            .concatMap(str -> Mono.subscriberContext()
                                  .flatMap(ctx -> ctx.<Mono<String>>get("someTopic"))
                                  .map(value -> str + value))
            .subscriberContext(Context.of(
                    "someTopic",
                    getValueFromDatabaseThatHaveReactiveInterface()
            ));

    StepVerifier.create(stringFlux)
                .expectNext("Hello hello")
                .expectNext("world hello")
                .verifyComplete();
}

private Mono<String> getValueFromDatabaseThatHaveReactiveInterface() {
    return Mono.just("hello");
}

In this example instead of trying to store the async value in the context, we may save the stream of values itself, and then, subscribe to that stream at any point in time we want. That technique is actively used in reactive spring-security (see and see) to resolve authenticated user from the storage

[apischan]

This actually doesn't work for my case. I will try to show more real world example where order of effects inside the chain have sense.

@Test
public void monoContext() {
    String message = "message";
    Flux<String> stringFlux = getTopicName(message)
            .flatMap(topic -> sendSomeRequestToTopic(topic).subscriberContext(ctx -> ctx.put("topic", topic)))
            .thenMany(getData())
            .concatMap(str -> Mono.subscriberContext().map(ctx -> str + ctx.get("topic"))
                    .flatMap(this::sendData)
            );

    StepVerifier.create(stringFlux)
            .expectNext("Hello hello")
            .expectNext("world hello")
            .verifyComplete();
}

/**
 * Getting the topic name from somewhere from database or from
 * something that have reactive interface
 */
private Mono<String> getTopicName(String ignore) {
    return Mono.just("hello");
}

/**
 * Emulating publishing the message to third party service
 * to prepare it to following requests
 */
private Mono<Void> sendSomeRequestToTopic(String topic) {
    return Mono.empty();
}

/**
 * Getting the data from the service that previously received
 * previous message from {@link this#sendSomeRequestToTopic()}
 */
private Flux<String> getData() {
    return Flux.just("Hello", "world");
}

/**
 * Send the modified data to some other service
 */
private Mono<String> sendData(String str) {
    return Mono.just(str);
}

[apischan]

The conceptual question: why "transferred from the downstream to the upstream at the subscriptions time" but not vice versa?

[simonbasle]

The Context was designed to provide some metadata to a Flux on a per-subscription basis, since a Flux by itself is just a description of a processing pipeline. The Subscription signal is already propagated from downstream to upstream.

Thus the Context is enriched during that phase. And as it is immutable values need to be "added" to it close to the final .subscribe() in order for them to be visible higher in the chain.

Keep in mind we need to keep compatibility with the Reactive Streams specification. Propagating mutations of the context in the data path (onNext) would have been a much larger deviation than just making it immutable and constructed in the subscription path.
Plus it would have had the potential for far worse performance overhead.

In short, Context is best used for providing metadata (like security credentials) from the subscription point, when said subscription is deferred elsewhere.

For your use-case, which seems to be more of an orchestration issue, I think it is better to make things explicit with intermediary Tuple2 (or a dedicated business POJO).

[OlegDokuka]

Hi again!

I guess mentioned example is not a use case for the context. In turn, I would rather solve that problem in the following way:

@Test
public void monoContext() {
    String message = "message";
    Flux<String> stringFlux = getTopicName(message)
            .flatMap(topic -> sendSomeRequestToTopic(topic)
                                  .thenMany(getData())
                                  .concatMap(str -> str + topic)
                                  .flatMap(this::sendData)
            );

    StepVerifier.create(stringFlux)
            .expectNext("Hello hello")
            .expectNext("world hello")
            .verifyComplete();
}

/**
 * Getting the topic name from somewhere from database or from
 * something that have reactive interface
 */
private Mono<String> getTopicName(String ignore) {
    return Mono.just("hello");
}

/**
 * Emulating publishing the message to third party service
 * to prepare it to following requests
 */
private Mono<Void> sendSomeRequestToTopic(String topic) {
    return Mono.empty();
}

/**
 * Getting the data from the service that previously received
 * previous message from {@link this#sendSomeRequestToTopic()}
 */
private Flux<String> getData() {
    return Flux.just("Hello", "world");
}

/**
 * Send the modified data to some other service
 */
private Mono<String> sendData(String str) {
    return Mono.just(str);
}

Regarding the second question, the safe way to fixate the state of the execution context for the whole stream is providing that context at the subscription time when there is no data, and the Execution flow transforms to the transformation flow. I guess it is similar to stack context that we have in Java. The final values for variables in the upper frame which is available for the newer frame are defined at the moment of the calling the lambda and by definition is effectively final (immutable). In streams world, the calling of the stream means subscribing to that stream. Hence the context is fixated at that moment and propagated from the point of subscription to the very beginning of that flow. (if it is still unclear, let me know and I will rephrase it)

[apischan]

Thanks for clarification.
The variant of @simonbasle in my case will work well but in more complicated cases will produce a lots of boilerplate. Just imagine instead of .thenMany(getData()) in my case more invocations like concatMap, map and so on. In this case I would be forced to rewrap value in all this invocations. And sometimes it would be even impossible.
The variant of @OlegDokuka as for me is better but it complicates the chain as well because of inner chains. But extracting inner chain to method will help.