Add Publisher#merge(...) operators

servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/Publisher.java

@@ -65,6 +65,7 @@
 import static io.servicetalk.concurrent.internal.SubscriberUtils.deliverErrorFromSource;
 import static io.servicetalk.utils.internal.DurationUtils.toNanos;
 import static java.util.Objects.requireNonNull;
+import static java.util.function.Function.identity;
 
 /**
  * An asynchronous computation that produces 0, 1 or more elements and may or may not terminate successfully or with
@@ -1521,6 +1522,78 @@ public final <R> Publisher<R> flatMapConcatIterable(Function<? super T, ? extend
         return new PublisherConcatMapIterable<>(this, mapper);
     }
 
+    /**
+     * Merge two {@link Publisher}s together. There is no guaranteed ordering of events emitted from the returned
+     * {@link Publisher}.
+     * <p>
+     * This method provides similar capabilities as expanding each result into a collection and concatenating each
+     * collection in sequential programming:
+     * <pre>{@code
+     *     List<T> mergedResults = ...; // concurrent safe list
+     *     for (T t : resultOfThisPublisher()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     for (T t : resultOfOtherPublisher()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     for (Future<R> future : futures) {
+     *         future.get(); // Throws if the processing for this item failed.
+     *     }
+     *     return mergedResults;
+     * }</pre>
+     * @param other The {@link Publisher} to merge with.
+     * @return A {@link Publisher} which is the result of this {@link Publisher} and {@code other} merged together.
+     * @see <a href="https://reactivex.io/documentation/operators/merge.html">ReactiveX merge operator</a>
+     * @see #mergeWithDelayError(Publisher)
+     * @see #merge(Publisher[])
+     */
+    public final Publisher<T> mergeWith(Publisher<T> other) {
+        return from(this, other).flatMapMerge(identity(), 2);
+    }
+
+    /**
+     * Merge two {@link Publisher}s together. There is no guaranteed ordering of events emitted from the returned
+     * {@link Publisher}. If either {@link Publisher} fails the error propagation will be delayed until both terminate.
+     * <p>
+     * This method provides similar capabilities as expanding each result into a collection and concatenating each
+     * collection in sequential programming:
+     * <pre>{@code
+     *     List<T> mergedResults = ...; // concurrent safe list
+     *     for (T t : resultOfThisPublisher()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     for (T t : resultOfOtherPublisher()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     List<Throwable> errors = ...;
+     *     for (Future<R> future : futures) {
+     *         try {
+     *           future.get(); // Throws if the processing for this item failed.
+     *         } catch (Throwable cause) {
+     *           errors.add(cause);
+     *         }
+     *     }
+     *     throwExceptionIfNotEmpty(errors);
+     *     return mergedResults;
+     * }</pre>
+     * @param other The {@link Publisher} to merge with,
+     * @return A {@link Publisher} which is the result of this {@link Publisher} and {@code other} merged together.
+     * @see <a href="https://reactivex.io/documentation/operators/merge.html">ReactiveX merge operator</a>
+     * @see #mergeWith(Publisher)
+     * @see #mergeDelayError(Publisher[])
+     */
+    public final Publisher<T> mergeWithDelayError(Publisher<T> other) {
+        return from(this, other).flatMapMergeDelayError(identity(), 2, 2);
+    }
+
     /**
      * Invokes the {@code onSubscribe} {@link Consumer} argument when
      * {@link Subscriber#onSubscribe(PublisherSource.Subscription)} is called for {@link Subscriber}s of the returned
@@ -4014,6 +4087,81 @@ public static <T> Publisher<T> defer(Supplier<? extends Publisher<? extends T>>
         return new PublisherDefer<>(publisherSupplier);
     }
 
+    /**
+     * Merge all {@link Publisher}s together. There is no guaranteed ordering of events emitted from the returned
+     * {@link Publisher}.
+     * <p>
+     * This method provides similar capabilities as expanding each result into a collection and concatenating each
+     * collection in sequential programming:
+     * <pre>{@code
+     *     List<T> mergedResults = ...; // concurrent safe list
+     *     for (T t : resultOfPublisher1()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     for (T t : resultOfOtherPublisher()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     for (Future<R> future : futures) {
+     *        future.get(); // Throws if the processing for this item failed.
+     *     }
+     *     return mergedResults;
+     * }</pre>
+     * @param publishers The {@link Publisher}s to merge together.
+     * @param <T> Type of items emitted by the returned {@link Publisher}.
+     * @return A {@link Publisher} which is the result of this {@link Publisher} and {@code other} merged together.
+     * @see <a href="https://reactivex.io/documentation/operators/merge.html">ReactiveX merge operator</a>
+     * @see #mergeDelayError(Publisher[])
+     */
+    @SafeVarargs
+    public static <T> Publisher<T> merge(Publisher<T>... publishers) {
+        return from(publishers).flatMapMerge(identity(), publishers.length);
+    }
+
+    /**
+     * Merge all {@link Publisher}s together. There is no guaranteed ordering of events emitted from the returned
+     * {@link Publisher}. If any {@link Publisher} terminates in an error, the error propagation will be delayed until
+     * all terminate.
+     * <p>
+     * This method provides similar capabilities as expanding each result into a collection and concatenating each
+     * collection in sequential programming:
+     * <pre>{@code
+     *     List<T> mergedResults = ...; // concurrent safe list
+     *     for (T t : resultOfPublisher1()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     for (T t : resultOfOtherPublisher()) {
+     *         futures.add(e.submit(() -> {
+     *             return mergedResults.add(t);
+     *         }));
+     *     }
+     *     List<Throwable> errors = ...;
+     *     for (Future<R> future : futures) {
+     *         try {
+     *           future.get(); // Throws if the processing for this item failed.
+     *         } catch (Throwable cause) {
+     *           errors.add(cause);
+     *         }
+     *     }
+     *     throwExceptionIfNotEmpty(errors);
+     *     return mergedResults;
+     * }</pre>
+     * @param publishers The {@link Publisher}s to merge together.
+     * @param <T> Type of items emitted by the returned {@link Publisher}.
+     * @return A {@link Publisher} which is the result of this {@link Publisher} and {@code other} merged together.
+     * @see <a href="https://reactivex.io/documentation/operators/merge.html">ReactiveX merge operator</a>
+     * @see #merge(Publisher[])
+     */
+    @SafeVarargs
+    public static <T> Publisher<T> mergeDelayError(Publisher<T>... publishers) {
+        return from(publishers).flatMapMergeDelayError(identity(), publishers.length, publishers.length);
+    }
+
     //
     // Static Utility Methods End
     //

servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/PublisherMergeWithTest.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright © 2023 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.concurrent.api;
+
+import io.servicetalk.concurrent.test.internal.TestPublisherSubscriber;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static io.servicetalk.concurrent.api.Publisher.merge;
+import static io.servicetalk.concurrent.api.Publisher.mergeDelayError;
+import static io.servicetalk.concurrent.api.SourceAdapters.toSource;
+import static io.servicetalk.concurrent.internal.DeliberateException.DELIBERATE_EXCEPTION;
+import static java.util.Arrays.asList;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.sameInstance;
+
+final class PublisherMergeWithTest {
+    private TestPublisher<Integer> first;
+    private TestPublisher<Integer> second;
+    private TestPublisher<Integer> third;
+    private TestPublisherSubscriber<Integer> subscriber;
+
+    @BeforeEach
+    void setUp() {
+        first = new TestPublisher<>();
+        second = new TestPublisher<>();
+        third = new TestPublisher<>();
+        subscriber = new TestPublisherSubscriber<>();
+    }
+
+    @SuppressWarnings("unused")
+    private static Iterable<Arguments> completeSource() {
+        List<Arguments> parameters = new ArrayList<>();
+        for (boolean inOrderOnNext : asList(false, true)) {
+            for (boolean inOrderTerminate : asList(false, true)) {
+                for (boolean firstOnError : asList(false, true)) {
+                    for (boolean delayError : asList(false, true)) {
+                        parameters.add(Arguments.of(inOrderOnNext, inOrderTerminate, firstOnError, delayError));
+                    }
+                }
+            }
+        }
+        return parameters;
+    }
+
+    @ParameterizedTest(name = "inOrderOnNext={0} inOrderTerminate={1} firstOnError={2} delayError={3}")
+    @MethodSource("completeSource")
+    void bothComplete(boolean inOrderOnNext, boolean inOrderTerminate, boolean firstOnError, boolean delayError) {
+        toSource(delayError ? first.mergeWithDelayError(second) : first.mergeWith(second)).subscribe(subscriber);
+        subscriber.awaitSubscription().request(2);
+        int i = 3;
+        int j = 4;
+        if (inOrderOnNext) {
+            first.onNext(i);
+            second.onNext(j);
+            assertThat(subscriber.takeOnNext(2), contains(i, j));
+        } else {
+            second.onNext(j);
+            first.onNext(i);
+            assertThat(subscriber.takeOnNext(2), contains(j, i));
+        }
+
+        if (inOrderTerminate) {
+            if (firstOnError) {
+                first.onError(DELIBERATE_EXCEPTION);
+
+                if (!delayError) {
+                    assertThat(subscriber.awaitOnError(), sameInstance(DELIBERATE_EXCEPTION));
+                }
+            } else {
+                first.onComplete();
+            }
+
+            second.onComplete();
+        } else {
+            if (firstOnError) {
+                second.onError(DELIBERATE_EXCEPTION);
+
+                if (!delayError) {
+                    assertThat(subscriber.awaitOnError(), sameInstance(DELIBERATE_EXCEPTION));
+                }
+            } else {
+                second.onComplete();
+            }
+            first.onComplete();
+        }
+
+        if (!firstOnError) {
+            subscriber.awaitOnComplete();
+        } else if (delayError) {
+            assertThat(subscriber.awaitOnError(), sameInstance(DELIBERATE_EXCEPTION));
+        }
+    }
+
+    @ParameterizedTest(name = "inOrderOnNext={0} inOrderTerminate={1} firstOnError={2} delayError={3}")
+    @MethodSource("completeSource")
+    void allComplete(boolean inOrderOnNext, boolean inOrderTerminate, boolean firstOnError, boolean delayError) {
+        toSource(delayError ? mergeDelayError(first, second, third) : merge(first, second, third))
+                .subscribe(subscriber);
+        subscriber.awaitSubscription().request(3);
+        int i = 3;
+        int j = 4;
+        int x = 5;
+        if (inOrderOnNext) {
+            first.onNext(i);
+            second.onNext(j);
+            third.onNext(x);
+            assertThat(subscriber.takeOnNext(3), contains(i, j, x));
+        } else {
+            second.onNext(j);
+            third.onNext(x);
+            first.onNext(i);
+            assertThat(subscriber.takeOnNext(3), contains(j, x, i));
+        }
+
+        if (inOrderTerminate) {
+            if (firstOnError) {
+                first.onError(DELIBERATE_EXCEPTION);
+
+                if (!delayError) {
+                    assertThat(subscriber.awaitOnError(), sameInstance(DELIBERATE_EXCEPTION));
+                }
+            } else {
+                first.onComplete();
+            }
+
+            second.onComplete();
+            third.onComplete();
+        } else {
+            if (firstOnError) {
+                third.onError(DELIBERATE_EXCEPTION);
+
+                if (!delayError) {
+                    assertThat(subscriber.awaitOnError(), sameInstance(DELIBERATE_EXCEPTION));
+                }
+            } else {
+                third.onComplete();
+            }
+            second.onComplete();
+            first.onComplete();
+        }
+
+        if (!firstOnError) {
+            subscriber.awaitOnComplete();
+        } else if (delayError) {
+            assertThat(subscriber.awaitOnError(), sameInstance(DELIBERATE_EXCEPTION));
+        }
+    }
+}

servicetalk-concurrent-reactivestreams/src/test/java/io/servicetalk/concurrent/reactivestreams/tck/PublisherZipWithDelayErrorTckTest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright © 2023 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.concurrent.reactivestreams.tck;
+
+import io.servicetalk.concurrent.api.Publisher;
+
+import static io.servicetalk.concurrent.api.Publisher.empty;
+
+public class PublisherZipWithDelayErrorTckTest extends AbstractPublisherOperatorTckTest<Integer> {
+    @Override
+    protected Publisher<Integer> composePublisher(final Publisher<Integer> publisher, final int elements) {
+        return publisher.mergeWithDelayError(empty());
+    }
+}

servicetalk-concurrent-reactivestreams/src/test/java/io/servicetalk/concurrent/reactivestreams/tck/PublisherZipWithTckTest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright © 2023 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.concurrent.reactivestreams.tck;
+
+import io.servicetalk.concurrent.api.Publisher;
+
+import static io.servicetalk.concurrent.api.Publisher.empty;
+
+public class PublisherZipWithTckTest extends AbstractPublisherOperatorTckTest<Integer> {
+    @Override
+    protected Publisher<Integer> composePublisher(final Publisher<Integer> publisher, final int elements) {
+        return publisher.mergeWith(empty());
+    }
+}