Better visibility on transport errors

servicetalk-concurrent-api-internal/src/main/java/io/servicetalk/concurrent/api/internal/SingleProcessor.java → servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/SingleProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2019 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.
@@ -13,10 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.servicetalk.concurrent.api.internal;
+package io.servicetalk.concurrent.api;
 
 import io.servicetalk.concurrent.Cancellable;
-import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.concurrent.internal.QueueFullAndRejectedSubscribeException;
 
 import java.util.Queue;
@@ -89,7 +88,7 @@ public void onSubscribe(final Cancellable cancellable) {
 
     @Override
     public void onSuccess(@Nullable final T result) {
-        terminate(result);
+        terminate(ThrowableWrapper.wrapIfThrowable(result));
     }
 
     @Override
@@ -109,10 +108,36 @@ private void notifyListeners(@Nullable Object terminalSignal) {
             drainSingleConsumerQueueDelayThrow(subscribers, subscriber -> subscriber.onError(error),
                     drainingTheQueueUpdater, this);
         } else {
-            @SuppressWarnings("unchecked")
-            final T value = (T) terminalSignal;
+            final T value = ThrowableWrapper.unwrapIfNeeded(terminalSignal);
             drainSingleConsumerQueueDelayThrow(subscribers, subscriber -> subscriber.onSuccess(value),
                     drainingTheQueueUpdater, this);
         }
     }
+
+    /**
+     * Wrapper to mark {@link Throwable} data so that we don't deliver it as {@link Subscriber#onError(Throwable)}.
+     */
+    private static final class ThrowableWrapper {
+        private final Throwable t;
+        private ThrowableWrapper(final Throwable t) {
+            this.t = t;
+        }
+
+        @SuppressWarnings("unchecked")
+        @Nullable
+        static <T> T unwrapIfNeeded(@Nullable Object inData) {
+            if (inData instanceof SingleProcessor.ThrowableWrapper) {
+                return (T) ((ThrowableWrapper) inData).t;
+            }
+            return (T) inData;
+        }
+
+        @Nullable
+        static Object wrapIfThrowable(@Nullable Object inData) {
+            if (inData instanceof Throwable) {
+                return new ThrowableWrapper((Throwable) inData);
+            }
+            return inData;
+        }
+    }
 }

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

@@ -0,0 +1,167 @@
+/*
+ * Copyright © 2019 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 org.junit.Rule;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+
+import static io.servicetalk.concurrent.internal.DeliberateException.DELIBERATE_EXCEPTION;
+
+public class SingleProcessorTest {
+    @Rule
+    public final MockedSingleListenerRule<?> rule = new MockedSingleListenerRule<>();
+    @Rule
+    public final MockedSingleListenerRule<?> rule2 = new MockedSingleListenerRule<>();
+
+    @SuppressWarnings("unchecked")
+    private <T> MockedSingleListenerRule<T> rule() {
+        return (MockedSingleListenerRule<T>) rule;
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T> MockedSingleListenerRule<T> rule2() {
+        return (MockedSingleListenerRule<T>) rule2;
+    }
+
+    @Test
+    public void testSuccessBeforeListen() {
+        testSuccessBeforeListen("foo");
+    }
+
+    @Test
+    public void testSuccessThrowableBeforeListen() {
+        testSuccessBeforeListen(DELIBERATE_EXCEPTION);
+    }
+
+    @Test
+    public void testSuccessNullBeforeListen() {
+        testSuccessBeforeListen(null);
+    }
+
+    private <T> void testSuccessBeforeListen(@Nullable T expected) {
+        SingleProcessor<T> processor = new SingleProcessor<>();
+        processor.onSuccess(expected);
+        rule().listen(processor).verifySuccess(expected);
+    }
+
+    @Test
+    public void testErrorBeforeListen() {
+        SingleProcessor<String> processor = new SingleProcessor<>();
+        processor.onError(DELIBERATE_EXCEPTION);
+        rule().listen(processor).verifyFailure(DELIBERATE_EXCEPTION);
+    }
+
+    @Test
+    public void testSuccessAfterListen() {
+        testSuccessAfterListen("foo");
+    }
+
+    @Test
+    public void testSuccessNullAfterListen() {
+        testSuccessAfterListen(null);
+    }
+
+    @Test
+    public void testSuccessThrowableAfterListen() {
+        testSuccessAfterListen(DELIBERATE_EXCEPTION);
+    }
+
+    private <T> void testSuccessAfterListen(@Nullable T expected) {
+        SingleProcessor<T> processor = new SingleProcessor<>();
+        rule().listen(processor).verifyNoEmissions();
+        processor.onSuccess(expected);
+        rule().verifySuccess(expected);
+    }
+
+    @Test
+    public void testErrorAfterListen() {
+        SingleProcessor<String> processor = new SingleProcessor<>();
+        rule().listen(processor).verifyNoEmissions();
+        processor.onError(DELIBERATE_EXCEPTION);
+        rule().verifyFailure(DELIBERATE_EXCEPTION);
+    }
+
+    @Test
+    public void testSuccessThenError() {
+        testSuccessThenError("foo");
+    }
+
+    @Test
+    public void testSuccessThrowableThenError() {
+        testSuccessThenError(DELIBERATE_EXCEPTION);
+    }
+
+    @Test
+    public void testSuccessNullThenError() {
+        testSuccessThenError(null);
+    }
+
+    private <T> void testSuccessThenError(@Nullable T expected) {
+        SingleProcessor<T> processor = new SingleProcessor<>();
+        processor.onSuccess(expected);
+        processor.onError(DELIBERATE_EXCEPTION);
+        rule().listen(processor).verifySuccess(expected);
+    }
+
+    @Test
+    public void testErrorThenSuccess() {
+        testErrorThenSuccess("foo");
+    }
+
+    @Test
+    public void testErrorThenSuccessThrowable() {
+        testErrorThenSuccess(DELIBERATE_EXCEPTION);
+    }
+
+    @Test
+    public void testErrorThenSuccessNull() {
+        testErrorThenSuccess(null);
+    }
+
+    private <T> void testErrorThenSuccess(@Nullable T expected) {
+        SingleProcessor<T> processor = new SingleProcessor<>();
+        processor.onError(DELIBERATE_EXCEPTION);
+        processor.onSuccess(expected);
+        rule().listen(processor).verifyFailure(DELIBERATE_EXCEPTION);
+    }
+
+    @Test
+    public void cancelRemovesListenerAndStillAllowsOtherListenersToBeNotified() {
+        cancelRemovesListenerAndStillAllowsOtherListenersToBeNotified("foo");
+    }
+
+    @Test
+    public void cancelRemovesListenerAndStillAllowsOtherListenersToBeNotifiedWithNull() {
+        cancelRemovesListenerAndStillAllowsOtherListenersToBeNotified(null);
+    }
+
+    @Test
+    public void cancelRemovesListenerAndStillAllowsOtherListenersToBeNotifiedWithThrowable() {
+        cancelRemovesListenerAndStillAllowsOtherListenersToBeNotified(DELIBERATE_EXCEPTION);
+    }
+
+    private <T> void cancelRemovesListenerAndStillAllowsOtherListenersToBeNotified(@Nullable T expected) {
+        SingleProcessor<T> processor = new SingleProcessor<>();
+        rule().listen(processor).verifyNoEmissions();
+        rule2().listen(processor).verifyNoEmissions();
+        rule().cancel();
+        processor.onSuccess(expected);
+        rule().verifyNoEmissions();
+        rule2().verifySuccess(expected);
+    }
+}

servicetalk-http-api/src/main/java/io/servicetalk/http/api/DefaultBlockingStreamingHttpRequest.java

@@ -21,7 +21,7 @@
 import io.servicetalk.concurrent.CloseableIterable;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
-import io.servicetalk.concurrent.api.internal.SingleProcessor;
+import io.servicetalk.concurrent.api.SingleProcessor;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpBufferFilterIterable;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpBuffersAndTrailersIterable;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpObjectsAndTrailersIterable;

servicetalk-http-api/src/main/java/io/servicetalk/http/api/DefaultBlockingStreamingHttpResponse.java

@@ -21,7 +21,7 @@
 import io.servicetalk.concurrent.CloseableIterable;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
-import io.servicetalk.concurrent.api.internal.SingleProcessor;
+import io.servicetalk.concurrent.api.SingleProcessor;
 import io.servicetalk.concurrent.internal.BlockingIterables;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpBufferFilterIterable;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpBuffersAndTrailersIterable;

servicetalk-http-api/src/main/java/io/servicetalk/http/api/DefaultStreamingHttpRequest.java

@@ -19,7 +19,7 @@
 import io.servicetalk.buffer.api.BufferAllocator;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
-import io.servicetalk.concurrent.api.internal.SingleProcessor;
+import io.servicetalk.concurrent.api.SingleProcessor;
 import io.servicetalk.http.api.HttpDataSourceTranformations.BridgeFlowControlAndDiscardOperator;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpBufferFilterOperator;
 import io.servicetalk.http.api.HttpDataSourceTranformations.HttpPayloadAndTrailersFromSingleOperator;