Single#concatPropagateCancel(Publisher) to force cancel propagation (#…

…2381)

Motivation:
In some scenarios concat is used where the right hand side has assocated state that is desirable to always trigger (concat payload body to meta data for http response processing). Regular concat will not subscribe if the Single terminates with onError or is cancelled which will prevent terminal event visibility.

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

@@ -764,9 +764,10 @@ public final Single<T> concat(Completable next) {
      * @return New {@link Publisher} that first emits the result of this {@link Single} and then subscribes and emits
      * all elements from {@code next} {@link Publisher}.
      * @see #concat(Publisher, boolean)
+     * @see #concatPropagateCancel(Publisher)
      */
     public final Publisher<T> concat(Publisher<? extends T> next) {
-        return new SingleConcatWithPublisher<>(this, next, false);
+        return concat(next, false);
     }
 
     /**
@@ -793,7 +794,29 @@ public final Publisher<T> concat(Publisher<? extends T> next) {
      * all elements from {@code next} {@link Publisher}.
      */
     public final Publisher<T> concat(Publisher<? extends T> next, boolean deferSubscribe) {
-        return new SingleConcatWithPublisher<>(this, next, deferSubscribe);
+        return new SingleConcatWithPublisher<>(this, next, deferSubscribe, false);
+    }
+
+    /**
+     * This method is like {@link #concat(Completable)} except {@code next} will be subscribed to and cancelled if this
+     * {@link Publisher} is cancelled or terminates with {@link Subscriber#onError(Throwable)}.
+     * <p>
+     * This method provides a means to sequence the execution of two asynchronous sources and in sequential programming
+     * is similar to:
+     * <pre>{@code
+     *     List<T> results = new ...;
+     *     results.add(resultOfThisSingle());
+     *     results.addAll(nextStream());
+     *     return results;
+     * }</pre>
+     * @param next {@link Publisher} to concat. Will be subscribed to and cancelled if this {@link Publisher} is
+     * cancelled or terminates with {@link Subscriber#onError(Throwable)}.
+     * @return New {@link Publisher} that first emits the result of this {@link Single} and then subscribes and emits
+     * all elements from {@code next} {@link Publisher}.
+     * @see #concat(Completable)
+     */
+    public final Publisher<T> concatPropagateCancel(Publisher<? extends T> next) {
+        return new SingleConcatWithPublisher<>(this, next, false, true);
     }
 
     /**