Save Object along the reactive pipeline.

[guy1699]

Hi, I wonder if the new Context Support can serve my need or if there is another tool.
During the pipeline some objects are created, then they need to be used at another point of the pipeline.
The important part is that each pipeline should have its own context (parallel to local thread) and this context should be available for the entire pipeline.
For example:

public Result someEndpoint() {
    return funcThatReturnUniAndSaveObjectOfTypeSessionOnTheContext(some parameters)
                      .flatMap(result -> doSomethingReturnUni(result))
                      .flatMap(result -> findOutIfThereIsSessionOnTheContextAndUseIt(result))
                      .flatMap(result -> findOutIfThereIsSessionOnTheContextAndUseItOneMoreTime(result))
} 

The subscribe is done by the platform (Spring-Boot), on whatever return from the endpoint.

[jponge]

As mentioned in #776, context is subscription-scoped, and it can certainly be updated when materialised, see https://github.com/smallrye/smallrye-mutiny/discussions

You just need to keep in mind that withContext operators are called at subscription time so not in the items flow order.
Because of that you should wrap the context manipulation operations using .onItem().invoke(...), etc.

If you don't pass a context at subscription time then an empty one will be available (not null!) so that might be helpful to you.

Note that I have no idea what your Spring Boot "platform" does regarding subscriptions; as far as I know Spring Boot does not support Mutiny but uses its own Reactor library.

[cescoffier]

I don't know much about the subscription handling of Spring Boot, but with Quarkus (and RESTEasy Reactive) you can do something like:

    AtomicInteger counter = new AtomicInteger();

    @GET
    @Path("/hey")
    public Uni<String> asyncGreeting(@RestQuery String name) {
        return Uni.createFrom().context(c -> {
            c.put("user.name", name);
            c.put("count", counter.incrementAndGet());
            return asyncGreeting();
        });
    }

    private Uni<String> asyncGreeting() {
        return Uni.createFrom()
                .context(c -> Uni.createFrom().item(() -> "Hey " + c.get("user.name") + " (" + c.get("count") + ")")
                .runSubscriptionOn(Infrastructure.getDefaultExecutor()));
    }

So, in your endpoint, you create a Uni with a context in which you pass what you need (here a query param and a count). In a later stage, you can retrieve the context.