cache() behaves unexpectedly with non-terminating sources

[chirayuk]

RxJS version: v5.0.0-beta.12

Code to reproduce:

var t$ = Rx.Observable.timer(0, 10).cache(1);
t$.take(5).subscribe();
// Wait until the previous subscription has ended.
// NOTE: t$ has NOT completed.
// So the following log should print one item from the cache and continue reading the rest from the timer.
// It should definitely NOT complete. But it does—prints a single item from the cache and completes.
setTimeout(() => t$.subscribe((x) => console.log(x)), 100);

Expected behavior:
Never ending console logs.

Actual behavior:
Single console log.

Additional information:
It's surprising behavior and undocumented (unless you read the code.) It definitely bit me and I'd consider it a bug.

[benlesh]

Looks like the issue is that when take() unsubscribes, it reduces the ref count to zero, and kills the source subscription

Doesn't seem like a hard fix... just requires a little thought.

[benlesh]

In fact, I think it requires thought to figure out if this is a bug, or just counter-intuitive behavior.

[benlesh]

Thinking about it, it seems like this isn't a bug, so much as it is counter-intuitive behavior.

consider:

const source = Observable.interval(1000).cache(1)

That part alone is a really strange use of cache, most likely... using it on an un-ending observable.

In the above case we're saying source is a ref-counted, hot observable. So it's not going to do anything until the first subscription to it. Since it's refCounted, we're going to stop the source when the subscriber count reaches zero.

If you use take(5) on it, and it completes before there are other subscribers to source, it's going to unsubscribe from the interval we're caching from.

Questions

Do we want a "refCounting" behavior with cache? Or shall we just say it goes until it completes (or errors)? What are the implications of that on other operations that could be performed? How does that affect our primary use case (which is mostly caching AJAX, I think)

[benlesh]

cc/ @jayphelps @trxcllnt @mattpodwysocki @jeffbcross @robwormald @kwonoj @staltz

[jayphelps]

If we can't reach consensus quickly, we might consider removing the operator entirely for v5.0 and try to get it (or something like it) into v5.1+ or some kitchen sink library.

The cache operator appears to have a history of confusion #839 #1541 #1940 #1765 which isn't surprising because an operator named "cache" is pretty ambiguous since there are so many strategies.

[chirayuk]

@Blesh—right on abut the counter-intuitive behavior.

My expectation is that the refCount going back to 0 when the source hasn't yet completed would still keep the subscription to the source until it actually terminates, and only then unsubscribes and get into the mode of returning cached behavior.

In my repro case, for instance, if I don't have the setTimeout in the last line, or it doesn't wait until the take() unsubscribes, then that 2nd subscribe never terminates.

While I used a never ending sequence, it doesn't have to be one, just one that terminates after take() has already unsubscribed.

[trxcllnt]

I honestly think the cache operator in its current form should be scrapped and re-thought entirely. Imo an operator that maintains a cache should abide by standard caching semantics (insertion, update, and invalidation), and would be memory-safe when the subscription to the cached Observable is closed.

Here's an example of what I mean:

import { Observable, ReplaySubject } from 'rxjs';

Observable.prototype.cache = function cache(
    keySelector,
    project,
    evictionSelector = () => Observable.never(),
    map = {}) {
    return this.groupBy(
        keySelector,
        (element) => element,
        (elements) => evictionSelector(elements)
            .ignoreElements()
            .finally(() => {
              console.log(`forgetting ${elements.key}`);
              delete map[elements.key];
            })
    )
    .mergeMap((elements) =>
        map[elements.key] || (
        map[elements.key] = project(elements)));
}

const cached = Observable
    .interval(100)
    .cache(
        (value) => value,
        (values) => {
            return values
                .multicast(new ReplaySubject())
                .refCount()
        }
    );

console.log('first subscribed');
cached
  .take(20)
  .subscribe(
    (i) => console.log(`first subscriber: ${i}`),
    (e) => console.error(e),
    ( ) => console.log('first complete')
  );

setTimeout(() => {
  console.log('second subscribed');
  cached
    .take(20)
    .subscribe(
      (i) => console.log(`second subscriber: ${i}`),
      (e) => console.error(e),
      ( ) => console.log('second complete')
    );
}, 1000);

[benlesh]

At this point, I'm in favor of pulling the cache() operator prior to a 5.0 release. This way we have time to reach a consensus what strategy we want to use around cache.

The goal was to provide an operator that could be used with AJAX use-cases that gave a simplistic caching behavior, similar to what one might find with promises, but providing the ability to abort the request, and the ability to retry the request.

Any other use case that might fall under "caching" wasn't really in this operator's design, however I can understand the confusion.

Alternatively, we could rename it to something like preserve...

[benlesh]

cache no longer exists. LOL.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.