Why UniMemoize doesn't call cancel on currentUpstreamSubscription after remove last awaiters in MemoizedSubscription.cancel?

[anton-tregubov]

Hi, please help me to understand. why MemoizedSubscription have so simple cancel method.

        public void cancel() {
            awaiters.remove(subscriber);
        }

In case when cancel remove LAST awaites and cachedResult still not filled better to cancel subscription on currentUpstreamSubscription . This is important to stop emitters and cleanup resources in memoized uni.

I try to implement same operator but with another MemoizedSubscription

            public void cancel()
            {
                internalLock.lock();
                try
                {
                    awaiters.remove(subscriber);
                    if (awaiters.isEmpty() && cachedResult == null && currentUpstreamSubscription != null)
                    {
                        state = State.INIT;
                        currentUpstreamSubscription.cancel();
                        currentUpstreamSubscription = null;
                    }
                }
                finally
                {
                    internalLock.unlock();
                }
            }

In my case it works fine. I do something wrong? Is this changes useful and can be merged to 2.x version (it allows to remove copy paste 90% line of code)?

[jponge]

Memoization "caches" a result for all the future subscribers. The contract is that the first subscription triggers some action that eventually yields a result, and that result becomes memoized until invalidated, so it's a simple state machine.

There's no guarantee that forwarding a cancellation will avoid receiving any value, as cancellation is never immediate.

Do you have any test / reproducer where the proposed implementation solves any correctness or performance issue?

[anton-tregubov]

I will try to simplify it. Cause in part another test.
My main problem following. I have huge evaluation logic that took some time.
And i memoize the result for this uni evaluation. This Uni share in several HTTP GET requests (if oth requests come when evaluation still in progress). Quarkus make several subscription for one memorized uni. And if all clients cancel requests i want to cleanup resource in upstream uni insted wait while this evaluation done. Cause new requests should start new evaluations from the beginnigs.

[g0rba4ev]

Also stumbled upon this behavior and this was unexpected and unintuitive.

The contract is that the first subscription triggers some action that eventually yields a result, and that result becomes memoized until invalidated, so it's a simple state machine.

@jponge is this statement documented somewhere?

---

Leaving my test case for others strangers

Uni<String> uni = Uni.createFrom().item("foo")
        .onItem().delayIt().by(Duration.ofSeconds(3))
        .log("upstreamUni")
        // if this memoization removed - upstream subscription is cancelled successfully
        .memoize().indefinitely(); 
        
Cancellable subscription = uni
        .log("downstreamUni")
        .subscribe().with(System.out::println);
        
Thread.sleep(Duration.ofSeconds(1));

subscription.cancel();

Thread.sleep(Duration.ofSeconds(10));

[jponge]

This is by design and that explanation is correct