fix: rewrite the concatMap operator

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.

implementation/revapi.json

@@ -51,7 +51,14 @@
     "criticality" : "highlight",
     "minSeverity" : "POTENTIALLY_BREAKING",
     "minCriticality" : "documented",
-    "differences" : [ ]
+    "differences" : [
+      {
+        "ignore": true,
+        "code": "java.class.removed",
+        "old": "class io.smallrye.mutiny.operators.multi.MultiConcatMapOp.ConcatMapMainSubscriber<I, O>",
+        "justification": "Refactoring of internal APIs"
+      }
+    ]
   }
 }, {
   "extension" : "revapi.reporter.json",

implementation/src/main/java/io/smallrye/mutiny/groups/MultiOnItem.java

@@ -264,7 +264,13 @@ public <O> Multi<O> transformToIterable(Function<? super T, ? extends Iterable<O
     @CheckReturnValue
     public <O> MultiFlatten<T, O> transformToUni(Function<? super T, Uni<? extends O>> mapper) {
         Function<? super T, Uni<? extends O>> actual = Infrastructure.decorate(nonNull(mapper, "mapper"));
-        Function<? super T, ? extends Publisher<? extends O>> wrapper = res -> actual.apply(res).toMulti();
+        Function<? super T, ? extends Publisher<? extends O>> wrapper = res -> {
+            Uni<? extends O> uni = actual.apply(res);
+            if (uni == null) {
+                return Multi.createFrom().failure(new NullPointerException(MAPPER_RETURNED_NULL));
+            }
+            return uni.toMulti();
+        };
         return new MultiFlatten<>(upstream, wrapper, 1, false);
     }