ConcatMap operator without prefetch

[ozangunalp]

Requests are forwarded lazily to the upstream when:

• First downstream request
• The inner has no more outstanding requests
• The inner completed without emitting items or with outstanding requests

Added entrypoints Multi.concatMap(mapper, prefetch) and MultiFlatten.concatanate(prefetch).
Existing transformTo[Uni|Multi]AndConcatanate methods default to prefetch=true for backwards compatible behavior.

Typically this operator allows applying async transformation on each item of a Multi, without prefetching an additional item from the upstream.

[codecov]

Codecov Report

Merging #997 (90a50f7) into main (1ddf9ec) will increase coverage by 0.15%.
The diff coverage is 79.54%.

@@             Coverage Diff              @@
##               main     #997      +/-   ##
============================================
+ Coverage     89.10%   89.26%   +0.15%     
- Complexity     3137     3150      +13     
============================================
  Files           412      413       +1     
  Lines         12566    12697     +131     
  Branches       1586     1617      +31     
============================================
+ Hits          11197    11334     +137     
+ Misses          717      708       -9     
- Partials        652      655       +3     

Impacted Files	Coverage Δ
...tation/src/main/java/io/smallrye/mutiny/Multi.java	94.11% <ø> (ø)
...llrye/mutiny/operators/multi/MultiConcatMapOp.java	78.74% <78.74%> (ø)
...n/java/io/smallrye/mutiny/groups/MultiFlatten.java	100.00% <100.00%> (ø)
...subscription/SwitchableSubscriptionSubscriber.java	97.61% <100.00%> (+2.41%)	⬆️
.../io/smallrye/mutiny/helpers/queues/DrainUtils.java	77.77% <0.00%> (-4.45%)	⬇️
...java/io/smallrye/mutiny/helpers/Subscriptions.java	79.00% <0.00%> (-1.66%)	⬇️
...ny/operators/uni/builders/UniCreateFromFuture.java	91.80% <0.00%> (-1.64%)	⬇️
...ny/operators/multi/builders/EmitterBasedMulti.java	94.33% <0.00%> (-0.95%)	⬇️
...main/kotlin/io/smallrye/mutiny/coroutines/Multi.kt	80.95% <0.00%> (ø)
...mallrye/mutiny/operators/multi/MultiFlatMapOp.java	90.03% <0.00%> (+0.33%)	⬆️
... and 11 more

[cescoffier]

@jponge

@ozangunalp and I discussed this yesterday. My proposal is the following:

• in 1.x, we keep the current default and concatMap prefetches by default. A new parameter can be passed (boolean) to enable/disable the prefetch. When disabled, we use the new operator.
• in 2.x, I was thinking of inverting the default, which means that by default, you will not have a prefetch. You would be able to switch back to the previous behavior by passing true to the method accepting the boolean parameter.

[cescoffier]

@jponge how do you want to proceed for the target branch? main and backport to 1.x, or 2 PRs? Note the discussion above that it may not be "just a backport", but a change of the default behavior.

[ozangunalp]

Ok, let me add a couple more test cases.
Meanwhile @jponge @cescoffier can you take a look?

[cescoffier]

The operator looks good.

[jponge]

@jponge how do you want to proceed for the target branch? main and backport to 1.x, or 2 PRs? Note the discussion above that it may not be "just a backport", but a change of the default behavior.

This PR is a feature / enhancement, so the target is 2.0.

If you want this in 1.x then yes you'll need a separate PR for the 1.x branch because it's not just a backport.

[jponge]

If there's a "backport" the target for it will be 1.7.0.

[ozangunalp]

To recap this PR targets the main branch (2.0) with concatenate methods defaulting to no prefetch.
Another PR will target 1.x branch for 1.7, keeping the default behaviour of prefetching on concatenate methods.

[jponge]

Thanks Ozan, I’ll get back to this next Monday.

…

On Fri, Jul 22, 2022, at 18:53, Ozan Gunalp wrote: To recap this PR targets the main branch (2.0) with `concatenate` methods defaulting to no prefetch. Another PR will target 1.x branch for 1.7, keeping the default behaviour of prefetching on `concatenate` methods. — Reply to this email directly, view it on GitHub <#997 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAAGK2I3YPT5SKZF5Y66VUTVVLG2NANCNFSM54HIDC6Q>. You are receiving this because you were mentioned.Message ID: ***@***.***>

[jponge]

LGTM, thanks @ozangunalp

[jponge]

Any further change or are we ready to merge?

[ozangunalp]

Thanks @jponge. It is ready to merge.