feat: improve type safety when combining unis with limited concurrency

implementation/revapi.json

@@ -51,7 +51,64 @@
     "criticality" : "highlight",
     "minSeverity" : "POTENTIALLY_BREAKING",
     "minCriticality" : "documented",
-    "differences" : [ ]
+    "differences" : [
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup2<T1, T2>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup2<T1, T2> io.smallrye.mutiny.groups.UniAndGroup2<T1, T2>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup2` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup3<T1, T2, T3>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup3<T1, T2, T3> io.smallrye.mutiny.groups.UniAndGroup3<T1, T2, T3>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup3` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup4<T1, T2, T3, T4>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup4<T1, T2, T3, T4> io.smallrye.mutiny.groups.UniAndGroup4<T1, T2, T3, T4>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup4` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup5<T1, T2, T3, T4, T5>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup5<T1, T2, T3, T4, T5> io.smallrye.mutiny.groups.UniAndGroup5<T1, T2, T3, T4, T5>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup5` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup6<T1, T2, T3, T4, T5, T6>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup6<T1, T2, T3, T4, T5, T6> io.smallrye.mutiny.groups.UniAndGroup6<T1, T2, T3, T4, T5, T6>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup6` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup7<T1, T2, T3, T4, T5, T6, T7>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup7<T1, T2, T3, T4, T5, T6, T7> io.smallrye.mutiny.groups.UniAndGroup7<T1, T2, T3, T4, T5, T6, T7>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup7` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup8<T1, T2, T3, T4, T5, T6, T7, T8>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup8<T1, T2, T3, T4, T5, T6, T7, T8> io.smallrye.mutiny.groups.UniAndGroup8<T1, T2, T3, T4, T5, T6, T7, T8>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup8` for improved type safety"
+      },
+      {
+        "ignore": true,
+        "code": "java.method.returnTypeChangedCovariantly",
+        "old": "method io.smallrye.mutiny.groups.UniAndGroupIterable<T1> io.smallrye.mutiny.groups.UniAndGroupIterable<T1>::usingConcurrencyOf(int) @ io.smallrye.mutiny.groups.UniAndGroup9<T1, T2, T3, T4, T5, T6, T7, T8, T9>",
+        "new": "method io.smallrye.mutiny.groups.UniAndGroup9<T1, T2, T3, T4, T5, T6, T7, T8, T9> io.smallrye.mutiny.groups.UniAndGroup9<T1, T2, T3, T4, T5, T6, T7, T8, T9>::usingConcurrencyOf(int)",
+        "justification": "Override `usingConcurrencyOf` in `UniAndGroup9` for improved type safety"
+      }
+    ]
   }
 }, {
   "extension" : "revapi.reporter.json",

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

@@ -47,6 +47,23 @@ public Uni<Tuple2<T1, T2>> asTuple() {
         return combine(Tuple2::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup2<T1, T2> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * Creates the resulting {@link Uni}. The items are combined using the given combinator function.
      *

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

@@ -30,6 +30,23 @@ public Uni<Tuple3<T1, T2, T3>> asTuple() {
         return combine(Tuple3::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup3<T1, T2, T3> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function3)} instead
      */

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

@@ -30,6 +30,23 @@ public Uni<Tuple4<T1, T2, T3, T4>> asTuple() {
         return combine(Tuple4::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup4<T1, T2, T3, T4> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function4)} instead
      */

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

@@ -31,6 +31,23 @@ public Uni<Tuple5<T1, T2, T3, T4, T5>> asTuple() {
         return combine(Tuple5::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup5<T1, T2, T3, T4, T5> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function5)} instead
      */

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

@@ -31,6 +31,23 @@ public Uni<Tuple6<T1, T2, T3, T4, T5, T6>> asTuple() {
         return combine(Tuple6::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup6<T1, T2, T3, T4, T5, T6> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function6)} instead
      */

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

@@ -31,6 +31,23 @@ public Uni<Tuple7<T1, T2, T3, T4, T5, T6, T7>> asTuple() {
         return combine(Tuple7::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup7<T1, T2, T3, T4, T5, T6, T7> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function7)} instead
      */

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

@@ -33,6 +33,23 @@ public Uni<Tuple8<T1, T2, T3, T4, T5, T6, T7, T8>> asTuple() {
         return combine(Tuple8::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup8<T1, T2, T3, T4, T5, T6, T7, T8> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function8)} instead
      */

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

@@ -33,6 +33,23 @@ public Uni<Tuple9<T1, T2, T3, T4, T5, T6, T7, T8, T9>> asTuple() {
         return combine(Tuple9::of);
     }
 
+    /**
+     * Limit the number of concurrent upstream subscriptions.
+     * <p>
+     * When not specified all upstream {@link Uni} are being subscribed when the combining {@link Uni} is subscribed.
+     * <p>
+     * Setting a limit is useful when you have a large number of {@link Uni} to combine and their simultaneous
+     * subscriptions might overwhelm resources (e.g., database connections, etc).
+     *
+     * @param level the concurrency level, must be strictly positive
+     * @return an object to configure the combination logic
+     */
+    @CheckReturnValue
+    public UniAndGroup9<T1, T2, T3, T4, T5, T6, T7, T8, T9> usingConcurrencyOf(int level) {
+        super.usingConcurrencyOf(level);
+        return this;
+    }
+
     /**
      * @deprecated use {@link #with(Functions.Function9)} instead
      */

implementation/src/test/java/io/smallrye/mutiny/operators/UniZipTest.java

@@ -636,6 +636,138 @@ void combineAllSmokeTest() {
             assertThat(sub.getItem()).asInstanceOf(LIST).containsExactly(1, 2, 3, 4);
         }
 
+        @Test
+        void combineAllTypesafeWith2() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2) -> Integer.parseInt(x1) + x2)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(3);
+        }
+
+        @Test
+        void combineAllTypesafeWith3() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3) -> Integer.parseInt(x1) + x2 + x3)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(6);
+        }
+
+        @Test
+        void combineAllTypesafeWith4() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+            Uni<Integer> d = Uni.createFrom().item(4);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c, d)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3, x4) -> Integer.parseInt(x1) + x2 + x3 + x4)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(10);
+        }
+
+        @Test
+        void combineAllTypesafeWith5() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+            Uni<Integer> d = Uni.createFrom().item(4);
+            Uni<Integer> e = Uni.createFrom().item(5);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c, d, e)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3, x4, x5) -> Integer.parseInt(x1) + x2 + x3 + x4 + x5)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(15);
+        }
+
+        @Test
+        void combineAllTypesafeWith6() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+            Uni<Integer> d = Uni.createFrom().item(4);
+            Uni<Integer> e = Uni.createFrom().item(5);
+            Uni<Integer> f = Uni.createFrom().item(6);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c, d, e, f)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3, x4, x5, x6) -> Integer.parseInt(x1) + x2 + x3 + x4 + x5 + x6)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(21);
+        }
+
+        @Test
+        void combineAllTypesafeWith7() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+            Uni<Integer> d = Uni.createFrom().item(4);
+            Uni<Integer> e = Uni.createFrom().item(5);
+            Uni<Integer> f = Uni.createFrom().item(6);
+            Uni<Integer> g = Uni.createFrom().item(7);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c, d, e, f, g)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3, x4, x5, x6, x7) -> Integer.parseInt(x1) + x2 + x3 + x4 + x5 + x6 + x7)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(28);
+        }
+
+        @Test
+        void combineAllTypesafeWith8() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+            Uni<Integer> d = Uni.createFrom().item(4);
+            Uni<Integer> e = Uni.createFrom().item(5);
+            Uni<Integer> f = Uni.createFrom().item(6);
+            Uni<Integer> g = Uni.createFrom().item(7);
+            Uni<Integer> h = Uni.createFrom().item(8);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c, d, e, f, g, h)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3, x4, x5, x6, x7, x8) -> Integer.parseInt(x1) + x2 + x3 + x4 + x5 + x6 + x7 + x8)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(36);
+        }
+
+        @Test
+        void combineAllTypesafeWith9() {
+            Uni<String> a = Uni.createFrom().item("1");
+            Uni<Integer> b = Uni.createFrom().item(2);
+            Uni<Integer> c = Uni.createFrom().item(3);
+            Uni<Integer> d = Uni.createFrom().item(4);
+            Uni<Integer> e = Uni.createFrom().item(5);
+            Uni<Integer> f = Uni.createFrom().item(6);
+            Uni<Integer> g = Uni.createFrom().item(7);
+            Uni<Integer> h = Uni.createFrom().item(8);
+            Uni<Integer> i = Uni.createFrom().item(9);
+
+            UniAssertSubscriber<Integer> sub = Uni.combine().all().unis(a, b, c, d, e, f, g, h, i)
+                    .usingConcurrencyOf(1)
+                    .with((x1, x2, x3, x4, x5, x6, x7, x8, x9) -> Integer.parseInt(x1) + x2 + x3 + x4 + x5 + x6 + x7 + x8 + x9)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+
+            sub.awaitItem().assertItem(45);
+        }
+
         @ParameterizedTest
         @CsvSource({
                 "1, 100, 1, 400",