cache behaving unexpected

[Dorus]

const source = Rx.Observable.merge(
      Observable.interval(100),
      Observable.of(42).do(() => console.log("  effect A"))
  );

const source2 = Observable.of(52).do(() => console.log("  effect B"));
const o1 = source.cache(1);
const o2 = source2.cache(1);
const ob = (name) => ({next: x => console.log(name + x), complete: () => console.log(name + "c")})
o1.take(3).subscribe(ob("a1 "));
o2.take(1).subscribe(ob("a2 "));
setTimeout(() => {
  o1.take(3).subscribe(ob("b1 "));
  o2.take(1).subscribe(ob("b2 "));
}, 500);

// setTimeout(() => {
//   o1.take(3).subscribe(ob("c1 "));
//   o2.take(1).subscribe(ob("c2 "));
// }, 1000);

// setTimeout(() => {
//   o1.take(3).subscribe(ob("d1 "));
//   o2.take(1).subscribe(ob("d2 "));
// }, 1500);

result in:

"  effect A"
"a1 42"
"  effect B"
"a2 52"
"a2 c"
"a1 0"
"a1 1"
"a1 c"
"b1 1"
"  effect A"
"b1 42"
"b2 52"
"b2 c"
"b1 0"
"b1 c"

What causes effect B to only play once, but effect A to play twice? Playing around with the code randomly seems to cause side-effects to appear and disappear on the source.

Because cache works like publishReplay().refCount(), i would expect effect B to also play again after a2 c (a2 complete), but b2 does not seem to query the source again.

On a possibly unrelated note, cache combines poorly with cold observables (or startWith) because it replays later values and then subscribes again. But this behavior is rather random because of the bug above.

[Dorus]

Also, an even worse example is

o1.take(2).subscribe(ob("a1 "));

setTimeout(() => {
  o1.take(2).subscribe(ob("b1 "));
}, 500);

setTimeout(() => {
  o1.take(2).subscribe(ob("c1 "));
}, 1000);

setTimeout(() => {
  o1.take(2).subscribe(ob("d1 "));
}, 1500);

where effect A plays 3 times, for a1, b1 and d1, but not for c1.

[trxcllnt]

This is working as designed. Your first source Observable (that logs "effect A") is combined with a Observable.interval() which never completes, whereas your second Observable (that logs "effect B") is a ScalarObservable that does complete.

When you create the first refCount subscriber to source1, you take three values, which unsubscribes from the source at the take location, and calls complete on the subscriber after the take. Critically, the source Observable (interval(100).concat(of(42))) never sent a complete message to the ReplaySubject created by cache(1). When you create the second refCount subscriber to source1, the ReplaySubject plays you back the most recent message, and since the ReplaySubject was never sent a complete message, is re-subscribed to the source Observable (interval(100).concat(of(42)).

This is in contrast to source2, which emits a single value (52) and then completes, which puts the ReplaySubject created by the second cache(1) call into a finalized state. Future refCount subscribers to o2 will be subscribed to the ReplaySubject, notified of the last value, then completed, but the ReplaySubject won't be re-subscribed to the source2. This is why you don't see "effect B" logged again.

[ronzeidman]

Hi,
look at this: https://jsbin.com/pozaho/4/edit?js,console

console.clear();

const {Observable} = Rx;

const observable = Observable.create(observer => {
  console.log('effect');
  observer.next('1');
  return () => {
    console.log('disposed');
  }
}).cache(1);

let subscription1 = observable.subscribe(val => console.log(`subscription1: ${val}`));
setTimeout(() => {
  subscription1.unsubscribe();
  subscription2 = observable.subscribe(val => console.log(`subscription2: ${val}`));
}, 500)

the output is this:

"effect"
"subscription1: 1"
"disposed"
"subscription2: 1"
"effect"
"subscription2: 1"

My issue is that I expect that if an observable was disposed the next subscription will act like the first one (= the cache would be cleared). Maybe this is not the design but is there a way to achieve this effect?
I need it for a very common scenario: I have a value that changed on the server and emits the changed value, upon subscription it emits the current value. I don't want the server to emit the current value for every subscription so I cache and share the observable. but if everyone unsubscribes and then someone subscribes he receives the cached value first which causes issues with further manipulations of the data (I aggregate/scan the server values to create an array out of the server's values so adding the cached value might adds itself to the resulting array)

[ronzeidman]

Also I have places where I need to get the latest value in a one-time matter (.take(1)) and I don't care if there are other subscribers or not. In that scenario I will always get the cached value even when there are no active subscription, thus I'd get an old value, and if I use .take(2) on the scenario where there are other subscribers I might not unsubscribe for a very long while unnecessarily.

[ronzeidman]

3rd scenario where the current cache behavior is bad - a security issue. When someone logs out of my site I unsubscribe from all observables, but when I log in with anther user I get the latest values that were relevant to the old user even though if the observable was reset the server would have known to send the right values due to the different token.

To sum up:
I use the cache so I will not have to subscribe directly to the server changes multiple times, this causes multiple issues all caused by the fact that cache sends the latest value even if the original observable was disposed:
1 - Values that are aggregated might get additional values when re-subscribed.
2 - Getting the latest up-to-date value is not trivial
3 - Getting other user's cached values when logging in after the other user logged out.

[Dorus]

@trxcllnt

What you say is false, i can add .concat(Rx.Observable.never) to the second observable and it will still only print effect B once. Also i used merge, not concat.

But beside that, my complain is not about the behavior of cache (whatever it might be), but about the unreliable behavior it shows. In my second example, you see i subscribe exactly the same 4x in a row, but it does different things from time to time. (Namely not printing the effect for c1).

If you are unable to reproduce this from my code sample, i can share the JsBin i used to test this in.

[trxcllnt]

@ronzeidman My issue is that I expect that if an observable was disposed the next subscription will act like the first one (= the cache would be cleared).

The behavior you're describing is what happens when you refCount a multicast source with a Subject selector function. In your example, if you change cache(1) to multicast(() => new ReplaySubject()).refCount(), this is the console output:

effect
subscription1: 1
disposed
effect
subscription2: 1

@Dorus I have submitted PR #1727 to address this issue, thanks for pointing it out.

[ronzeidman]

@trxcllnt Thanks! sorry for the issue hijack :) multicast(() => new ReplaySubject(1)).refCount() works perfectly for my scenarios. Thanks also for fixing the original cache issue, I also see that the multicast approach does not suffer the same unreliability that the cache originally suffered from.

[benlesh]

This should be resolved with #1727

[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.