Context support open for review

[jponge]

I just assembled my work on Mutiny contexts in #776 and I'd be happy to get feedbacks!

Here's a brief overview.

What's a context?

This PR offers subscriber-provided contexts, so all operators involved in a subscription can share some form of implicit data.

A context is a simple key / value, in-memory storage. Data can be queried, added and deleted from a context, as shown in the following snippet:

// Create a context using key / value pairs
Context context = Context.of(
        "X-CUSTOMER-ID", "1234",
        "X-SPAN-ID", "foo-bar-baz"
);

// Get an entry
System.out.println(
        context.<String>get("X-SPAN-ID"));

// Get an entry, use a supplier for a default value if the key is not present
System.out.println(
        context.getOrElse("X-SPAN-ID", () -> "<no id>"));

// Add an entry
context.put("foo", "bar");

// Remove an entry
context.delete("foo");

Where can I see some examples?

You can see the full tests suite at: https://github.com/smallrye/smallrye-mutiny/blob/feature/context-support/implementation/src/test/java/io/smallrye/mutiny/ContextTest.java

Here's a quick snippet that shall (hopefully) be self-explanatory:

// Context with 1 entry
Context context = Context.of("X-CUSTOMER-ID", customerId);

// "pipeline" is a Multi
pipeline.withContext((multi, ctx) -> multi.onItem().transformToUniAndMerge(item -> makeRequest(item, ctx.get("X-CUSTOMER-ID"))))
        .subscribe().with(context, item -> handleResponse(item), err -> handleFailure(err));

[gavinking]

oh. very interesting.

[jponge]

If it's a Uni from a CompletionStage then (Mutiny) context shall propagate.

In the other cases no, MP context propagation will be a better option.

[jponge]

For the record something (convoluted) like this works:

Context context = Context.of("foo", "bar");

Uni.createFrom().completionStage(() -> supplyAsync(() -> "Yolo"))
    .withContext((uni, ctx) -> uni.onItem().transformToUni(str -> Uni.createFrom().completionStage(() -> supplyAsync(() -> str + "::" + ctx.get("foo")))))
    .onItem().transformToUni(str -> Uni.createFrom().completionStage(() -> supplyAsync(() -> str)))
    .subscribeAsCompletionStage(context)
    .whenComplete((str, err) -> System.out.println(str));

Note that the CompletableFuture instances don't get automagic access to the context.

[gavinking]

Well the completion stages don't need the context. What I care about is mutiny Unis that are on either side of the intervening chain of completion stages.

[jponge]

Yes the intent is that the context is available between the subscriber and all Uni/Multi in a chain.

[gavinking]

So that sounds good then.

[FroMage]

What's the use-case? Most JEE stuff that needs context will never go through the pain of passing it as an extra parameter, which is why they're in thread-locals for decades.

[jponge]

It's been on the roadmap of discussions we've had with @cescoffier for a long time. The use-case is indeed to support some form of implicit data/state inside a Mutiny pipeline, much like thread-locals do.

Without some form of context you need to explicitly pass state (think of tuples), so this brings an abstraction for that.
It's not meant to be a replacement for other forms of contexts, but one that matches the Mutiny / reactive streams model.

[jponge]

Without any further comment in the PR by tomorrow morning my time I'll go ahead with the merge.

[nilyin]

Hi, can you please provide an example of how to use context in combination with Mutiny session for Hibernate-reactive? My use-case is to store results of multiple database queries and fill the needed DTO at the end of my reactive chain of queries for that session. something like this:

`return sessionFactory.withTransaction((session,t) -> session.find(Entity.class, id)
.onItem().ifNull().failWith(() -> { throw new NotFoundException("failed to find Entity"); })
.onItem().ifNotNull().transformToUni(entity ->
session.createQuery("select eo.id from EntityOption eo where eo.entity =: myEntity", Long.class).setParameter("myEntity", entity).getResultList()
/*
// here I need to store the 1st query' result to get it later at the end of pipeline, so it should look like:
.onItem().invoke(list -> ctx.put("options_list", list)
.chain (res -> next query) and so on

// as I understand using "invoke" is preferable here (compared to "call") as it guarantees that my list is saved in the context's cache before next chained query happens, at it blocks the pipeline, right?
*/

my another wish is that the session context should be automatically cleared / flushed at the end of the transaction.
BTW, may be Mutiny session should have its own built-in context? like session.getContext().put(k,v) or session.put(k,v)