Fail-proof UniMemoizeOp subscriber removal
@@ -130,6 +130,7 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
case CACHING:
wrapper.subscriber.onSubscribe(wrapper::markCancelled);
wrapper.markSubscribed();
+ try {
if (!wrapper.isCancelled()) {
if (currentFailure != null) {
wrapper.subscriber.onFailure(currentFailure);
@@ -137,7 +138,9 @@ public class UniMemoizeOp<I> extends UniOperator<I, I> implements UniSubscriber<
wrapper.subscriber.onItem(currentItem);
}
}
+ } finally {
subscribers.remove(wrapper);
+ }
break;
default:
throw new IllegalStateException("Current state is " + state);
