fix: rewrite the concatMap operator

[jponge]

In this implementation the operator acts as a simple no-queue forwarder
between the subscriber and the current upstream.

There is also a small dose of fine-grained locking around a few
state machine updates that can’t be expressed as non-blocking / compare & swap
operations.

[jponge]

@ozangunalp How does this sound against the main branch of Reactive Messaging?

[codecov]

Codecov Report

Attention: 15 lines in your changes are missing coverage. Please review.

Comparison is base (82adadc) 89.23% compared to head (dded2d6) 89.34%.
Report is 1 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #1489      +/-   ##
============================================
+ Coverage     89.23%   89.34%   +0.10%     
- Complexity     3362     3365       +3     
============================================
  Files           459      459              
  Lines         13405    13434      +29     
  Branches       1642     1635       -7     
============================================
+ Hits          11962    12002      +40     
- Misses          805      808       +3     
+ Partials        638      624      -14     

Files	Coverage Δ
...in/java/io/smallrye/mutiny/groups/MultiOnItem.java	100.00% <100.00%> (ø)
...llrye/mutiny/operators/multi/MultiConcatMapOp.java	87.33% <87.80%> (+4.93%)	⬆️

... and 14 files with indirect coverage changes

[ozangunalp]

@jponge Reactive Messaging tests are green 🥇
Looking at the code now

[ozangunalp]

It seems all correct to me.
Only weird thing to me is some updates to the state are done with CAS field updater others with the lock.

[jponge]

It seems all correct to me. Only weird thing to me is some updates to the state are done with CAS field updater others with the lock.

Thanks Ozan!

Yes some updates can be done in a CAS, but in some cases you need to read the current state but update depending on some other condition (e.g., is there any outstanding demand or not), which IMHO can't be expressed as a CAS.

[ozangunalp]

but in some cases you need to read the current state but update depending on some other condition

Exactly that!

[jponge]

@ozangunalp May I ask you one more run? (I did a few edits, nothing that should break, but those are always famous last words)

[ozangunalp]

I've checked again. Reactive messaging tests in local are green.

[cescoffier]

didn't notice anything weird.

[jponge]

didn't notice anything weird.

... "Yet" 🤣