Single#concat(Publisher) onNext error propagation (#1615)

Motivation:
Single#concat(Publisher) doesn't catch exceptions from downstream
Subscriber#onNext and relies upon upstream to propagate exceptions.
However upstream is a Single and has already terminated by the time the
exception is caught so it cannot send a duplicate terminal signal.

Modifications:
- Single#concat(Publisher) catches exceptions from downstream
  Subscriber#onNext and propagates an error.

Result:
Single#concat(Publisher) is more robust with respect to exception
handling.

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

@@ -152,7 +152,12 @@ public void cancel() {
         }
 
         private void emitSingleSuccessToTarget(@Nullable final T result) {
-            target.onNext(result);
+            try {
+                target.onNext(result);
+            } catch (Throwable cause) {
+                target.onError(cause);
+                return;
+            }
             next.subscribeInternal(this);
         }
     }

servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/single/SingleConcatWithPublisherTest.java

@@ -56,6 +56,19 @@ void setUp() {
         subscriber.awaitSubscription();
     }
 
+    @Test
+    void onNextErrorPropagated() {
+        subscriber = new TestPublisherSubscriber<>();
+        source = new TestSingle<>();
+        toSource(source.concat(next).<Integer>map(x -> {
+            throw DELIBERATE_EXCEPTION;
+        })).subscribe(subscriber);
+        subscriber.awaitSubscription().request(1);
+        source.onSuccess(1);
+        assertThat(subscriber.awaitOnError(), is(DELIBERATE_EXCEPTION));
+        assertThat(next.isSubscribed(), is(false));
+    }
+
     @Test
     void bothCompletion() {
         triggerNextSubscribe();