Context support

[jponge]

This brings subscriber-provided contexts to Uni and Multi.

Context can be materialized using the withContext and attachContext operators.

See #757

This is a squashed commit based on explorations in https://github.com/jponge/smallrye-mutiny/tree/feature/context

[jponge]

• Compatibility checks on Quarkus
• Compatibility checks on Hibernate Reactive
• Performance checks
• Can we find a way to materialize the context in builders? (so far only emitters can do that)

[Sanne]

do you really need them to be both volatile and concurrent?

[jponge]

In theory we could have 2 concurrent put that'd cause the initialisation

[Sanne]

Yes I understand the need with the current design; but I'm curious about

1. why the whole Context needs to be concurrent-safe (even for initialization and writes)
2. why you opted for lazy initialization of the underlying map (I do love the attention towards allocations but I wonder how likely this is to not be allocated - and if it's very low, getting rid of this could simplify such code)
3. did you consider immutable contexts? Mutations could be modelled as a new view over an existing context (admittedly this would be less flexible and only suitable for relatively small collections but it could be fine - so it's a deeper question about intended use)

[jponge]

1. Since you have 1 Context instance for a whole subscription lifespan, there are cases where events and subscriptions get offset to an executor. Or you merge pipelines (e.g., 2 concurrent HTTP requests), etc. There is 0 guarantee we won't have concurrent access for such kind of object, hence IMHO it's better to make it concurrent-proof from the start.
2. The code is more elegant / safer with non-null Context objects around, for instance some subscribers provide an empty context rather than flying a null around. Deferring the allocation is indeed all about reducing the footprint: subscription that use a context will allocate, those who don't will not pay the full price.
3. My first iterations had been around read / write views, but I found the developer experience to be worse than it should be. First off immutable contexts are nice on the paper, but you need to remember that the binding methods are called at subscription time, which is counter-intuitive (events flow from publishers to subscribers, but here contexts flow from subscribers to publishers). I found out that have withContext (and attachContext as sugar) to bind was simple and flexible (no need to patch tons of methods) and has clear semantics. Then I found out that there was little value in read/write views, if you are after manipulating a context across a whole pipeline then making it a shared, thread-safe object was more intuitive.

(hope it makes sense, lots of points in a single post 😄 )

[Sanne]

Many thanks for all thourough explanations! Makes sense.

[jponge]

Marking as ready for review, but in any case let's have this PR bake a little bit.

[jponge]

Added {Uni,Multi}.createFrom().context(ctx -> pipeline), as in:

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

AssertSubscriber<String> sub = Multi.createFrom().context(ctx -> Multi.createBy()
        .repeating().uni(
                AtomicInteger::new,
                counter -> Uni.createFrom()
                        .item(counter.incrementAndGet() + "::" + ctx.getOrElse("foo", () -> "yolo")))
        .atMost(5))
        .subscribe().withSubscriber(AssertSubscriber.create(context, Long.MAX_VALUE));

sub.assertCompleted();
assertThat(sub.getItems())
        .hasSize(5)
        .contains("1::bar", "5::bar");

and:

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

Multi<String> a = Multi.createFrom()
        .context(ctx -> Multi.createFrom().items("a", ctx.getOrElse("foo", () -> "yolo")));

Multi<String> b = Multi.createFrom()
        .context(ctx -> Multi.createFrom().items("b", ctx.getOrElse("foo", () -> "yolo")));

Multi<String> c = Multi.createFrom()
        .context(ctx -> Multi.createFrom().items("c", ctx.getOrElse("foo", () -> "yolo")));

AssertSubscriber<String> sub = Multi.createBy().merging().streams(a, b, c)
        .subscribe().withSubscriber(AssertSubscriber.create(context, Long.MAX_VALUE));

sub.assertCompleted();
assertThat(sub.getItems())
        .hasSize(6)
        .containsExactly("a", "bar", "b", "bar", "c", "bar");

[codecov]

Codecov Report

Merging #776 (267f168) into main (b83b5db) will increase coverage by 0.02%.
The diff coverage is 80.06%.

❗ Current head 267f168 differs from pull request most recent head f02bc50. Consider uploading reports for the commit f02bc50 to get more accurate results

@@             Coverage Diff              @@
##               main     #776      +/-   ##
============================================
+ Coverage     89.10%   89.13%   +0.02%     
- Complexity     2987     3081      +94     
============================================
  Files           377      384       +7     
  Lines         11891    12161     +270     
  Branches       1500     1525      +25     
============================================
+ Hits          10596    10840     +244     
- Misses          682      700      +18     
- Partials        613      621       +8     

Impacted Files	Coverage Δ
...smallrye/mutiny/helpers/MultiEmitterProcessor.java	90.19% <0.00%> (-1.81%)	⬇️
.../operators/multi/MultiSelectFirstUntilOtherOp.java	78.43% <0.00%> (-1.57%)	⬇️
.../mutiny/operators/multi/MultiSkipUntilOtherOp.java	89.13% <0.00%> (-1.99%)	⬇️
...mutiny/operators/multi/builders/ResourceMulti.java	84.46% <0.00%> (-2.54%)	⬇️
...o/smallrye/mutiny/subscription/ContextSupport.java	0.00% <0.00%> (ø)
...entation/src/main/java/io/smallrye/mutiny/Uni.java	88.23% <25.00%> (-8.44%)	⬇️
...mutiny/operators/multi/MultiOperatorProcessor.java	87.27% <33.33%> (-3.12%)	⬇️
...iny/operators/multi/builders/BaseMultiEmitter.java	80.39% <33.33%> (-2.95%)	⬇️
...erators/multi/builders/SerializedMultiEmitter.java	80.48% <33.33%> (-0.53%)	⬇️
...utiny/operators/uni/UniOnItemTransformToMulti.java	84.00% <33.33%> (-3.24%)	⬇️
... and 59 more

[jponge]

Codecov reports are wrong (it reports untested code that's actually being tested).

[jponge]

@cescoffier Fancy a review to celebrate 2022? 😉

[cescoffier]

LGTM,

We would need feedback on the usage (so, this feature will be marked as experimental for a bit).

[jponge]

@cescoffier No remark on docs / etc? Yes all new APIs have been marked as experimental.

[cescoffier]

No, it looks ok. However, I would need to try it to be sure it does what I believe it does :-D

Typically, I'm not totally sure it could be used for the Multi serialization use case we have in reactive routes, where the Publisher must extend the context and this context would be read in the subscriber (to apply the right serialization).

[jponge]

Typically, I'm not totally sure it could be used for the Multi serialization use case we have in reactive routes, where the Publisher must extend the context and this context would be read in the subscriber (to apply the right serialization).

Let me know if you need some delay before we merge this one.

[jponge]

Last call for comments, or I'll merge tomorrow (Tuesday 4th) morning my time 😃

[jponge]

...and of course I need to rebase and fix a conflict.

[cescoffier]

Go ahead!

…

On 3 Jan 2022, at 20:24, Julien Ponge ***@***.***> wrote:  Last call for comments, or I'll merge tomorrow (Tuesday 4th) morning my time 😃 — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you were mentioned.

[jponge]

It’s available in Mutiny 1.3.0

…

On 13 Jan 2022, at 12:43, Guy Assaf ***@***.***> wrote: @jponge <https://github.com/jponge> How can I consume it in my application, is it available as Maven dependency? — Reply to this email directly, view it on GitHub <#776 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAAGK2M4VZSGDZZXZUXEK3DUV23GHANCNFSM5J7G2JTA>. You are receiving this because you were mentioned.

[guy1699]

Thanks @jponge, found it, an amazing job.
I'm working on keeping the current Hibernate session/transaction along the same reactive stream.

I don't understand how to use the context,
can you give an example other than the above?

In this example, I debug and the code start from "number 3 " last reactive chain, why?
Also, the context doesn't keep the value along the stream.

return Uni.createFrom().item(tenant1).withContext((someEntity, context) -> {
                    context.put("UUID", UUID.randomUUID());
                    LOGGER.info("number 1 UUID=" + context.get("UUID"));
                    return someEntity;
                })
                .map(someEntity -> someEntity)
                .withContext((someEntity, context) -> {
                    LOGGER.info("number 2 UUID=" + context.get("UUID"));
                    return someEntity;
                })
                .map(someEntity -> someEntity)
                .withContext((someEntity, context) -> {

                    LOGGER.info("number 3 UUID=" + context.get("UUID"));
                    return someEntity;
                }).map(tenant ->
                        tenant);

[jponge]

The withContext method is being called at subscription time, so it goes from the subscription to the source (also see https://smallrye.io/smallrye-mutiny/guides/context-passing)

Instead of manipulating the context from inside the lambda, you should perform the manipulations from an operator, as in:

.withContext((someEntity, context) ->
  someEntity.onItem().invoke(() -> LOGGER.info("number 3 UUID=" + context.get("UUID")) )

(not sure it compiles, it's a sketch)

Does that make sense?

[guy1699]

So where do you manipulate (PUT) the object into the Context?
It needs to happen during the pipeline only then I have the object, I'm not triggering the subscription, it is done by Spring-boot on the entire endpoint.
@jponge please assist :)

We can move this conversation to a better place if this is not the right location.

[jponge]

You can perform Context manipulation operations in various event processing handlers (e.g., onItem().invoke(item -> { ... ctx.put(foo, bar) }, etc.

In general you should pass all context data that you know at subscription time (hint: I have no idea what happens in your case) and then you can read/write/delete in the context-aware portions of your pipeline.

If you have further questions you can start a discussion at https://github.com/smallrye/smallrye-mutiny/discussions

[guy1699]

Thanks I will open a discussion.