Memoization cleanups
limit calls to the drain loop
avoid unnecessary awaiting subscription list addition/removal on early exit
@@ -46,7 +46,6 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
@Override
protected void subscribing(UniSerializedSubscriber<? super I> subscriber) {
- awaitingSubscription.add(subscriber);
if (invalidationRequested.getAsBoolean() && state.get() != State.SUBSCRIBING) {
state.set(State.INIT);
if (upstreamSubscription != null) {
@@ -54,8 +53,8 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
}
}
+ // Early exit with cached data
if (state.get() == State.CACHING) {
- awaitingSubscription.remove(subscriber);
subscriber.onSubscribe(EmptyUniSubscription.CANCELLED);
if (failure != null) {
subscriber.onFailure(failure);
@@ -65,20 +64,20 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
return;
}
+ awaitingSubscription.add(subscriber);
if (state.compareAndSet(State.INIT, State.SUBSCRIBING)) {
// This thread is performing the upstream subscription
AbstractUni.subscribe(upstream(), this);
}
- drain();
}
@Override
public void onSubscribe(UniSubscription subscription) {
if (state.compareAndSet(State.SUBSCRIBING, State.SUBSCRIBED)) {
upstreamSubscription = subscription;
- }
drain();
}
+ }
private void drain() {
// Check if another thread is working
@@ -169,9 +168,9 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
this.item = item;
this.failure = null;
state.set(State.CACHING);
- }
drain();
}
+ }
@Override
public void onFailure(Throwable failure) {
@@ -179,7 +178,7 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
this.item = null;
this.failure = failure;
state.set(State.CACHING);
- }
drain();
}
}
+}
