SourceAdapters should convert back and forth without allocation (#3196)

Motivation:

`SourceAdapters.toSource(...)` currently returns a source without
implementing the original type. In result, operation like
`fromSource(toSource(notSource))` returns double wrapped object instead
of wrapping only once.

Modifications:

- `PublisherToPublisherSource` extends `Publisher`;
- `SingleToSingleSource` extends `Single`;
- `CompletableToCompletableSource` extends `Completable`;
- Enhance `SourceAdaptersTest` to cover all transformations (via cast
and via wrapping).

Result:

Less wrapping layers for back-and-forth conversions and better test
coverage for `SourceAdapters`.

Author: idelpivnitskiy

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

@@ -27,7 +27,11 @@ abstract class AbstractNoHandleSubscribeCompletable extends SubscribableCompleta
 
     @Override
     protected final void handleSubscribe(final Subscriber subscriber) {
-        deliverErrorFromSource(subscriber, new UnsupportedOperationException("Subscribe with no " +
-                CapturedContext.class.getSimpleName() + " is not supported for " + getClass()));
+        deliverErrorFromSource(subscriber, newUnsupportedOperationException(getClass()));
+    }
+
+    static UnsupportedOperationException newUnsupportedOperationException(final Class<?> clazz) {
+        return new UnsupportedOperationException(
+                "Subscribe with no " + CapturedContext.class.getSimpleName() + " is not supported for " + clazz);
     }
 }

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

@@ -17,6 +17,7 @@
 
 import io.servicetalk.concurrent.api.SubscribableSources.SubscribablePublisher;
 
+import static io.servicetalk.concurrent.api.AbstractNoHandleSubscribeCompletable.newUnsupportedOperationException;
 import static io.servicetalk.concurrent.internal.SubscriberUtils.deliverErrorFromSource;
 
 /**
@@ -29,7 +30,6 @@ abstract class AbstractNoHandleSubscribePublisher<T> extends SubscribablePublish
 
     @Override
     protected final void handleSubscribe(final Subscriber<? super T> subscriber) {
-        deliverErrorFromSource(subscriber, new UnsupportedOperationException("Subscribe with no " +
-                CapturedContext.class.getSimpleName() + " is not supported for " + getClass()));
+        deliverErrorFromSource(subscriber, newUnsupportedOperationException(getClass()));
     }
 }

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

@@ -17,6 +17,7 @@
 
 import io.servicetalk.concurrent.api.SubscribableSources.SubscribableSingle;
 
+import static io.servicetalk.concurrent.api.AbstractNoHandleSubscribeCompletable.newUnsupportedOperationException;
 import static io.servicetalk.concurrent.internal.SubscriberUtils.deliverErrorFromSource;
 
 /**
@@ -29,7 +30,6 @@ abstract class AbstractNoHandleSubscribeSingle<T> extends SubscribableSingle<T>
 
     @Override
     protected final void handleSubscribe(final Subscriber<? super T> subscriber) {
-        deliverErrorFromSource(subscriber, new UnsupportedOperationException("Subscribe with no " +
-                CapturedContext.class.getSimpleName() + " is not supported for " + getClass()));
+        deliverErrorFromSource(subscriber, newUnsupportedOperationException(getClass()));
     }
 }

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

@@ -39,6 +39,6 @@ void handleSubscribe(final Subscriber subscriber,
         // This operator currently only targets the subscribe method. Given this limitation if we try to change the
         // ContextMap now it is possible that operators downstream in the subscribe call stack may have modified
         // the ContextMap and we don't want to discard those changes by using a different ContextMap.
-        original.handleSubscribe(subscriber, capturedContext, contextProvider);
+        original.delegateSubscribe(subscriber, capturedContext, contextProvider);
     }
 }

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

@@ -33,6 +33,6 @@ void handleSubscribe(final Subscriber subscriber,
         // This operator currently only targets the subscribe method. Given this limitation if we try to change the
         // ContextMap now it is possible that operators downstream in the subscribe call stack may have modified
         // the ContextMap and we don't want to discard those changes by using a different ContextMap.
-        original.handleSubscribe(subscriber, capturedContext, contextProvider);
+        original.delegateSubscribe(subscriber, capturedContext, contextProvider);
     }
 }

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

@@ -39,6 +39,6 @@ void handleSubscribe(final Subscriber<? super T> singleSubscriber,
         // This operator currently only targets the subscribe method. Given this limitation if we try to change the
         // ContextMap now it is possible that operators downstream in the subscribe call stack may have modified
         // the ContextMap and we don't want to discard those changes by using a different ContextMap.
-        original.handleSubscribe(singleSubscriber, capturedContext, contextProvider);
+        original.delegateSubscribe(singleSubscriber, capturedContext, contextProvider);
     }
 }

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

@@ -33,6 +33,6 @@ void handleSubscribe(final Subscriber<? super T> singleSubscriber,
         // This operator currently only targets the subscribe method. Given this limitation if we try to change the
         // ContextMap now it is possible that operators downstream in the subscribe call stack may have modified
         // the ContextMap and we don't want to discard those changes by using a different ContextMap.
-        original.handleSubscribe(singleSubscriber, capturedContext, contextProvider);
+        original.delegateSubscribe(singleSubscriber, capturedContext, contextProvider);
     }
 }

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

@@ -39,6 +39,6 @@ void handleSubscribe(final Subscriber<? super T> singleSubscriber,
         // This operator currently only targets the subscribe method. Given this limitation if we try to change the
         // ContextMap now it is possible that operators downstream in the subscribe call stack may have modified
         // the ContextMap and we don't want to discard those changes by using a different ContextMap.
-        original.handleSubscribe(singleSubscriber, capturedContext, contextProvider);
+        original.delegateSubscribe(singleSubscriber, capturedContext, contextProvider);
     }
 }

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

@@ -33,6 +33,6 @@ void handleSubscribe(final Subscriber<? super T> singleSubscriber,
         // This operator currently only targets the subscribe method. Given this limitation if we try to change the
         // ContextMap now it is possible that operators downstream in the subscribe call stack may have modified
         // the ContextMap and we don't want to discard those changes by using a different ContextMap.
-        original.handleSubscribe(singleSubscriber, capturedContext, contextProvider);
+        original.delegateSubscribe(singleSubscriber, capturedContext, contextProvider);
     }
 }

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

@@ -19,6 +19,8 @@
 import io.servicetalk.concurrent.PublisherSource;
 import io.servicetalk.concurrent.SingleSource;
 
+import static io.servicetalk.concurrent.api.AbstractNoHandleSubscribeCompletable.newUnsupportedOperationException;
+import static io.servicetalk.concurrent.internal.SubscriberUtils.deliverErrorFromSource;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -133,39 +135,75 @@ private static <T> SingleSource<T> uncheckedCast(final Single<T> single) {
         return (SingleSource<T>) single;
     }
 
-    private static final class PublisherToPublisherSource<T> implements PublisherSource<T> {
+    // Visible for testing
+    static final class PublisherToPublisherSource<T> extends Publisher<T> implements PublisherSource<T> {
         private final Publisher<T> publisher;
 
         PublisherToPublisherSource(final Publisher<T> publisher) {
             this.publisher = requireNonNull(publisher);
         }
 
+        @Override
+        protected void handleSubscribe(final Subscriber<? super T> subscriber) {
+            deliverErrorFromSource(subscriber, newUnsupportedOperationException(getClass()));
+        }
+
+        @Override
+        void handleSubscribe(final Subscriber<? super T> subscriber,
+                             final CapturedContext capturedContext, final AsyncContextProvider contextProvider) {
+            publisher.delegateSubscribe(subscriber, capturedContext, contextProvider);
+        }
+
         @Override
         public void subscribe(final Subscriber<? super T> subscriber) {
             publisher.subscribeInternal(subscriber);
         }
     }
 
-    private static final class SingleToSingleSource<T> implements SingleSource<T> {
+    // Visible for testing
+    static final class SingleToSingleSource<T> extends Single<T> implements SingleSource<T> {
         private final Single<T> single;
 
         SingleToSingleSource(final Single<T> single) {
             this.single = requireNonNull(single);
         }
 
+        @Override
+        protected void handleSubscribe(final Subscriber<? super T> subscriber) {
+            deliverErrorFromSource(subscriber, newUnsupportedOperationException(getClass()));
+        }
+
+        @Override
+        void handleSubscribe(final Subscriber<? super T> subscriber,
+                             final CapturedContext capturedContext, final AsyncContextProvider contextProvider) {
+            single.delegateSubscribe(subscriber, capturedContext, contextProvider);
+        }
+
         @Override
         public void subscribe(final Subscriber<? super T> subscriber) {
             single.subscribeInternal(subscriber);
         }
     }
 
-    private static final class CompletableToCompletableSource implements CompletableSource {
+    // Visible for testing
+    static final class CompletableToCompletableSource extends Completable implements CompletableSource {
         private final Completable completable;
 
         CompletableToCompletableSource(final Completable completable) {
             this.completable = requireNonNull(completable);
         }
 
+        @Override
+        protected void handleSubscribe(final Subscriber subscriber) {
+            deliverErrorFromSource(subscriber, newUnsupportedOperationException(getClass()));
+        }
+
+        @Override
+        void handleSubscribe(final Subscriber subscriber,
+                             final CapturedContext capturedContext, final AsyncContextProvider contextProvider) {
+            completable.delegateSubscribe(subscriber, capturedContext, contextProvider);
+        }
+
         @Override
         public void subscribe(final Subscriber subscriber) {
             completable.subscribeInternal(subscriber);
@@ -180,7 +218,7 @@ private static final class PublisherSourceToPublisher<T> extends Publisher<T> im
         }
 
         @Override
-        protected void handleSubscribe(final PublisherSource.Subscriber<? super T> subscriber) {
+        protected void handleSubscribe(final Subscriber<? super T> subscriber) {
             source.subscribe(subscriber);
         }
 
@@ -198,7 +236,7 @@ private static final class SingleSourceToSingle<T> extends Single<T> implements
         }
 
         @Override
-        protected void handleSubscribe(final SingleSource.Subscriber<? super T> subscriber) {
+        protected void handleSubscribe(final Subscriber<? super T> subscriber) {
             source.subscribe(subscriber);
         }
 
@@ -216,7 +254,7 @@ private static final class CompletableSourceToCompletable extends Completable im
         }
 
         @Override
-        protected void handleSubscribe(final CompletableSource.Subscriber subscriber) {
+        protected void handleSubscribe(final Subscriber subscriber) {
             source.subscribe(subscriber);
         }