Merge pull request #379 from smallrye/internal/uni-refactor-to-UniSub…

…scriber
smallrye · Dec 8, 2020 · 832e918 · 832e918
2 parents 2615d01 + cd2831e
commit 832e918
Show file tree

Hide file tree

Showing 47 changed files with 189 additions and 99 deletions.
diff --git a/...ropagation/src/main/java/io/smallrye/mutiny/context/ContextPropagationUniInterceptor.java b/...ropagation/src/main/java/io/smallrye/mutiny/context/ContextPropagationUniInterceptor.java
@@ -7,7 +7,6 @@
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.infrastructure.UniInterceptor;
 import io.smallrye.mutiny.operators.AbstractUni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
 import io.smallrye.mutiny.subscription.UniSubscriber;
 import io.smallrye.mutiny.subscription.UniSubscription;
 
@@ -44,7 +43,7 @@ public <T> Uni<T> onUniCreation(Uni<T> uni) {
         Executor executor = getThreadContext().currentContextExecutor();
         return new AbstractUni<T>() {
             @Override
-            protected void subscribing(UniSerializedSubscriber<? super T> subscriber) {
+            protected void subscribing(UniSubscriber<? super T> subscriber) {
                 executor.execute(() -> AbstractUni.subscribe(uni, subscriber));
             }
         };

diff --git a/implementation/revapi.json b/implementation/revapi.json
@@ -14,6 +14,7 @@
             "io\\.smallrye\\.mutiny\\..*"
           ],
           "exclude": [
+            "io\\.smallrye\\.mutiny\\.operators\\.*",
             "io\\.smallrye\\.mutiny\\.operators\\..*"
           ]
         }
@@ -32,6 +33,54 @@
           "old": "method <T> io.smallrye.mutiny.helpers.Subscriptions.SingleItemSubscription<T> io.smallrye.mutiny.helpers.Subscriptions::single(org.reactivestreams.Subscriber<T>, T)",
           "new": "method <T> org.reactivestreams.Subscription io.smallrye.mutiny.helpers.Subscriptions::single(org.reactivestreams.Subscriber<T>, T)",
           "justification": "SingleItemSubscription should not be exposed, and should be used as a regular Subscription"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniGlobalSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniGlobalSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniOnCancellationSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniOnCancellationSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniOnFailureSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniOnFailureSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniOnItemOrFailureSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniOnItemOrFailureSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniOnItemSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniOnItemSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniOnSubscribeSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniOnSubscribeSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.helpers.spies.UniOnTerminationSpy<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.helpers.spies.UniOnTerminationSpy<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
+        },
+        {
+          "code": "java.method.parameterTypeChanged",
+          "old": "parameter void io.smallrye.mutiny.operators.AbstractUni<T>::subscribing(===io.smallrye.mutiny.operators.UniSerializedSubscriber<? super T>===)",
+          "new": "parameter void io.smallrye.mutiny.operators.AbstractUni<T>::subscribing(===io.smallrye.mutiny.subscription.UniSubscriber<? super T>===)",
+          "justification": "subscribing(subscriber) is a protected method that is not used by regular consumers of the Mutiny public API"
         }
       ]
     }

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniGlobalSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniGlobalSpy.java
@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.helpers.spies;
 
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniGlobalSpy<T> extends UniSpyBase<T> {
 
@@ -68,7 +68,7 @@ public void reset() {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         onTerminationSpy.subscribe().withSubscriber(downstream);
     }
 

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnCancellationSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnCancellationSpy.java
@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.helpers.spies;
 
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniOnCancellationSpy<T> extends UniSpyBase<T> {
 
@@ -10,7 +10,7 @@ public class UniOnCancellationSpy<T> extends UniSpyBase<T> {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         upstream()
                 .onCancellation().invoke(this::incrementInvocationCount)
                 .subscribe().withSubscriber(downstream);

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnFailureSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnFailureSpy.java
@@ -4,7 +4,7 @@
 
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.groups.UniOnFailure;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniOnFailureSpy<T> extends UniSpyBase<T> {
 
@@ -38,7 +38,7 @@ public void reset() {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         UniOnFailure<? extends T> group;
         if (predicate != null) {
             group = upstream().onFailure(predicate);

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnItemOrFailureSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnItemOrFailureSpy.java
@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.helpers.spies;
 
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniOnItemOrFailureSpy<T> extends UniSpyBase<T> {
 
@@ -32,7 +32,7 @@ public Throwable lastFailure() {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         upstream()
                 .onItemOrFailure().invoke((item, failure) -> {
                     synchronized (UniOnItemOrFailureSpy.this) {

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnItemSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnItemSpy.java
@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.helpers.spies;
 
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniOnItemSpy<T> extends UniSpyBase<T> {
 
@@ -22,7 +22,7 @@ public void reset() {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         upstream()
                 .onItem().invoke(item -> {
                     this.lastItem = item;

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnSubscribeSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnSubscribeSpy.java
@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.helpers.spies;
 
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 import io.smallrye.mutiny.subscription.UniSubscription;
 
 public class UniOnSubscribeSpy<T> extends UniSpyBase<T> {
@@ -23,7 +23,7 @@ public void reset() {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         upstream()
                 .onSubscribe().invoke(uniSubscription -> {
                     incrementInvocationCount();

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnTerminationSpy.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/spies/UniOnTerminationSpy.java
@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.helpers.spies;
 
 import io.smallrye.mutiny.Uni;
-import io.smallrye.mutiny.operators.UniSerializedSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 import io.smallrye.mutiny.tuples.Tuple3;
 
 public class UniOnTerminationSpy<T> extends UniSpyBase<T> {
@@ -31,7 +31,7 @@ public void reset() {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> downstream) {
+    protected void subscribing(UniSubscriber<? super T> downstream) {
         upstream()
                 .onTermination().invoke((i, f, c) -> {
                     incrementInvocationCount();

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/AbstractUni.java b/implementation/src/main/java/io/smallrye/mutiny/operators/AbstractUni.java
@@ -13,7 +13,7 @@
 
 public abstract class AbstractUni<T> implements Uni<T> {
 
-    protected abstract void subscribing(UniSerializedSubscriber<? super T> subscriber);
+    protected abstract void subscribing(UniSubscriber<? super T> subscriber);
 
     /**
      * Encapsulates subscription to slightly optimized the AbstractUni case.

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniAndCombination.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniAndCombination.java
@@ -41,7 +41,7 @@ public UniAndCombination(Uni<? extends I> upstream, List<? extends Uni<?>> other
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super O> subscriber) {
+    protected void subscribing(UniSubscriber<? super O> subscriber) {
         AndSupervisor andSupervisor = new AndSupervisor(subscriber);
         subscriber.onSubscribe(andSupervisor);
         // Must wait until the subscriber get a subscription before subscribing to the sources.
@@ -51,11 +51,11 @@ protected void subscribing(UniSerializedSubscriber<? super O> subscriber) {
     private class AndSupervisor implements UniSubscription {
 
         private final List<UniHandler> handlers = new ArrayList<>();
-        private final UniSerializedSubscriber<? super O> subscriber;
+        private final UniSubscriber<? super O> subscriber;
 
         AtomicBoolean cancelled = new AtomicBoolean();
 
-        AndSupervisor(UniSerializedSubscriber<? super O> sub) {
+        AndSupervisor(UniSubscriber<? super O> sub) {
             subscriber = sub;
 
             for (Uni u : unis) {

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateFromCompletionStage.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateFromCompletionStage.java
@@ -7,6 +7,7 @@
 import java.util.function.Supplier;
 
 import io.smallrye.mutiny.helpers.ParameterValidation;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniCreateFromCompletionStage<O> extends UniOperator<Void, O> {
     private final Supplier<? extends CompletionStage<? extends O>> supplier;
@@ -17,7 +18,7 @@ public UniCreateFromCompletionStage(Supplier<? extends CompletionStage<? extends
     }
 
     private static <O> void forwardFromCompletionStage(CompletionStage<? extends O> stage,
-            UniSerializedSubscriber<? super O> subscriber) {
+            UniSubscriber<? super O> subscriber) {
         subscriber.onSubscribe(() -> stage.toCompletableFuture().cancel(false));
         stage.whenComplete((res, fail) -> {
             if (fail != null) {
@@ -33,7 +34,7 @@ private static <O> void forwardFromCompletionStage(CompletionStage<? extends O>
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super O> subscriber) {
+    protected void subscribing(UniSubscriber<? super O> subscriber) {
         CompletionStage<? extends O> stage;
         try {
             stage = supplier.get();

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateFromDeferredSupplier.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateFromDeferredSupplier.java
@@ -7,6 +7,7 @@
 
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.helpers.ParameterValidation;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniCreateFromDeferredSupplier<T> extends UniOperator<Void, T> {
     private final Supplier<Uni<? extends T>> supplier;
@@ -17,7 +18,7 @@ public UniCreateFromDeferredSupplier(Supplier<Uni<? extends T>> supplier) {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> subscriber) {
+    protected void subscribing(UniSubscriber<? super T> subscriber) {
         nonNull(subscriber, "subscriber");
         Uni<? extends T> uni;
         try {

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateFromPublisher.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateFromPublisher.java
@@ -10,6 +10,7 @@
 import org.reactivestreams.Subscription;
 
 import io.smallrye.mutiny.infrastructure.Infrastructure;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniCreateFromPublisher<O> extends UniOperator<Void, O> {
     private final Publisher<? extends O> publisher;
@@ -21,7 +22,7 @@ public UniCreateFromPublisher(Publisher<? extends O> publisher) {
 
     @SuppressWarnings("SubscriberImplementation")
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super O> subscriber) {
+    protected void subscribing(UniSubscriber<? super O> subscriber) {
         AtomicReference<Subscription> reference = new AtomicReference<>();
         Subscriber<O> actual = new Subscriber<O>() {
             @Override

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateWithEmitter.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniCreateWithEmitter.java
@@ -5,6 +5,7 @@
 import java.util.function.Consumer;
 
 import io.smallrye.mutiny.subscription.UniEmitter;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniCreateWithEmitter<T> extends AbstractUni<T> {
     private final Consumer<UniEmitter<? super T>> consumer;
@@ -14,7 +15,7 @@ public UniCreateWithEmitter(Consumer<UniEmitter<? super T>> consumer) {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> subscriber) {
+    protected void subscribing(UniSubscriber<? super T> subscriber) {
         DefaultUniEmitter<? super T> emitter = new DefaultUniEmitter<>(subscriber);
         subscriber.onSubscribe(emitter);
 

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniDelayOnItem.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniDelayOnItem.java
@@ -11,6 +11,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 import io.smallrye.mutiny.subscription.UniSubscription;
 
 public class UniDelayOnItem<T> extends UniOperator<T, T> {
@@ -24,7 +25,7 @@ public UniDelayOnItem(Uni<T> upstream, Duration duration, ScheduledExecutorServi
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> subscriber) {
+    protected void subscribing(UniSubscriber<? super T> subscriber) {
         AtomicReference<ScheduledFuture<?>> holder = new AtomicReference<>();
         AtomicReference<UniSubscription> reference = new AtomicReference<>();
         AbstractUni.subscribe(upstream(), new UniDelegatingSubscriber<T, T>(subscriber) {

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniDelayUntil.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniDelayUntil.java
@@ -8,6 +8,7 @@
 import java.util.function.Function;
 
 import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 import io.smallrye.mutiny.subscription.UniSubscription;
 
 public class UniDelayUntil<T> extends UniOperator<T, T> {
@@ -22,7 +23,7 @@ public UniDelayUntil(Uni<T> upstream, Function<? super T, Uni<?>> function,
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super T> subscriber) {
+    protected void subscribing(UniSubscriber<? super T> subscriber) {
         AtomicReference<UniSubscription> reference = new AtomicReference<>();
         AbstractUni.subscribe(upstream(), new UniDelegatingSubscriber<T, T>(subscriber) {
             @Override

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniDelegatingSubscriber.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniDelegatingSubscriber.java
@@ -7,9 +7,9 @@
 
 public class UniDelegatingSubscriber<I, O> implements UniSubscriber<I> {
 
-    private final UniSerializedSubscriber<? super O> delegate;
+    private final UniSubscriber<? super O> delegate;
 
-    public UniDelegatingSubscriber(UniSerializedSubscriber<? super O> subscriber) {
+    public UniDelegatingSubscriber(UniSubscriber<? super O> subscriber) {
         this.delegate = nonNull(subscriber, "delegate");
     }
 

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniEmitOn.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniEmitOn.java
@@ -5,6 +5,7 @@
 import java.util.concurrent.Executor;
 
 import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.subscription.UniSubscriber;
 
 public class UniEmitOn<I> extends UniOperator<I, I> {
     private final Executor executor;
@@ -15,7 +16,7 @@ public class UniEmitOn<I> extends UniOperator<I, I> {
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super I> subscriber) {
+    protected void subscribing(UniSubscriber<? super I> subscriber) {
         AbstractUni.subscribe(upstream(), new UniDelegatingSubscriber<I, I>(subscriber) {
             @Override
             public void onItem(I item) {

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/UniFailOnTimeout.java b/implementation/src/main/java/io/smallrye/mutiny/operators/UniFailOnTimeout.java
@@ -33,7 +33,7 @@ public UniFailOnTimeout(Uni<I> upstream, Duration timeout, Supplier<? extends Th
     }
 
     @Override
-    protected void subscribing(UniSerializedSubscriber<? super I> subscriber) {
+    protected void subscribing(UniSubscriber<? super I> subscriber) {
         AtomicBoolean doneOrCancelled = new AtomicBoolean();
         AtomicReference<ScheduledFuture<?>> task = new AtomicReference<>();
 
@@ -92,7 +92,7 @@ public void onFailure(Throwable failure) {
         });
     }
 
-    private void sendTimeout(UniSerializedSubscriber<? super I> subscriber, UniSubscription subscription) {
+    private void sendTimeout(UniSubscriber<? super I> subscriber, UniSubscription subscription) {
 
         // Cancel the upstream subscription
         subscription.cancel();