Reduce FlatMapMainSubscriber footprint
@@ -73,7 +73,9 @@ public final class MultiFlatMapOp<I, O> extends AbstractMultiOperator<I, O> {
volatile boolean done;
volatile boolean cancelled;
- final AtomicReference<Subscription> upstream = new AtomicReference<>();
+ volatile Subscription upstream = null;
+ private static final AtomicReferenceFieldUpdater<FlatMapMainSubscriber, Subscription> UPSTREAM_UPDATER = AtomicReferenceFieldUpdater
+ .newUpdater(FlatMapMainSubscriber.class, Subscription.class, "upstream");
AtomicLong requested = new AtomicLong();
@@ -148,7 +150,7 @@ public final class MultiFlatMapOp<I, O> extends AbstractMultiOperator<I, O> {
if (wip.getAndIncrement() == 0) {
clearQueue();
- upstream.getAndSet(Subscriptions.CANCELLED).cancel();
+ UPSTREAM_UPDATER.getAndSet(this, Subscriptions.CANCELLED).cancel();
unsubscribe();
}
}
@@ -156,7 +158,7 @@ public final class MultiFlatMapOp<I, O> extends AbstractMultiOperator<I, O> {
@Override
public void onSubscribe(Subscription s) {
- if (upstream.compareAndSet(null, s)) {
+ if (UPSTREAM_UPDATER.compareAndSet(this, null, s)) {
downstream.onSubscribe(this);
s.request(Subscriptions.unboundedOrRequests(maxConcurrency));
}
@@ -445,7 +447,7 @@ public final class MultiFlatMapOp<I, O> extends AbstractMultiOperator<I, O> {
}
if (replenishMain != 0L && !done && !cancelled) {
- upstream.get().request(replenishMain);
+ upstream.request(replenishMain);
}
if (again) {
@@ -461,7 +463,7 @@ public final class MultiFlatMapOp<I, O> extends AbstractMultiOperator<I, O> {
private void cancelUpstream(boolean fromOnError) {
clearQueue();
- Subscription subscription = upstream.getAndSet(Subscriptions.CANCELLED);
+ Subscription subscription = UPSTREAM_UPDATER.getAndSet(this, Subscriptions.CANCELLED);
if (subscription != null) {
subscription.cancel();
}
