Use Mutiny signals rather than RS signals and extra cancellation checks
@@ -56,6 +56,9 @@ public class MultiDemandPacer<T> extends AbstractMultiOperator<T, T> {
}
private void demandAndSchedule(ScheduledExecutorService executor) {
+ if (upstream == Subscriptions.CANCELLED) {
+ return;
+ }
long demand = currentRequest.demand();
long delay = currentRequest.delay().toNanos();
scheduledFuture = executor.schedule(this::tick, delay, TimeUnit.NANOSECONDS);
@@ -84,7 +87,8 @@ public class MultiDemandPacer<T> extends AbstractMultiOperator<T, T> {
@Override
public void onSubscribe(Subscription subscription) {
- super.onSubscribe(subscription);
+ if (compareAndSetUpstreamSubscription(null, subscription)) {
+ downstream.onSubscribe(this);
try {
currentRequest = pacer.initial();
} catch (Throwable failure) {
@@ -98,6 +102,9 @@ public class MultiDemandPacer<T> extends AbstractMultiOperator<T, T> {
} else {
demandAndSchedule(executor);
}
+ } else {
+ subscription.cancel();
+ }
}
@Override
@@ -106,7 +113,7 @@ public class MultiDemandPacer<T> extends AbstractMultiOperator<T, T> {
return;
}
itemsCounter.incrementAndGet();
- downstream.onNext(item);
+ downstream.onItem(item);
}
@Override
@@ -115,7 +122,7 @@ public class MultiDemandPacer<T> extends AbstractMultiOperator<T, T> {
return;
}
cancel();
- downstream.onError(failure);
+ downstream.onFailure(failure);
}
@Override
@@ -124,7 +131,7 @@ public class MultiDemandPacer<T> extends AbstractMultiOperator<T, T> {
return;
}
cancel();
- downstream.onComplete();
+ downstream.onCompletion();
}
@Override
