Rxjs: Unsubscription doesn't work after shareReplay?!

[ivan-kleshnin]

RxJS version: 5.5.2

Code to reproduce:

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.min.js"></script>

<button id="foo">Foo</button>
<button id="bar">Bar</button>

<script>
  let {Observable: O} = Rx

  let fooNode = document.querySelector("#foo")
  let barNode = document.querySelector("#bar")

  let fooClick$ = O.fromEvent(fooNode, "click")
    .do(() => { console.log("foo clicked!") })

  let barClick$ = O.fromEvent(barNode, "click")
    .do(() => { console.log("bar clicked!") }).shareReplay(1) // !!! 

  let fooSb = fooClick$.subscribe()
  let barSb = barClick$.subscribe() // !!!

  setTimeout(() => {
    fooSb.unsubscribe()
  }, 2000)

  setTimeout(() => {
    barSb.unsubscribe() // !!!
  }, 2000)
</script>

Expected behavior:

Both buttons stop working after two seconds.

Actual behavior:

Only the first button stop working after two seconds.

[cartant]

I'm unable to reproduce this with the script that's in the issue.

Well after two seconds has elapsed, clicking on the Bar button effects the logging of bar clicked!. The subscription to the source observable does not appear to be unsubscribed when the subscription to the shared observable is unsubscribed.

[ivan-kleshnin]

Well after two seconds has elapsed, clicking on the Bar button effects the logging of bar clicked!.

Yes, that's a problem.

The subscription to the source observable does not appear to be unsubscribed when the subscription to the shared observable is unsubscribed.

Yes and I believe it's a bug because shareReplay now produces an unremovable event listener (which is inappropriate in any scenario).

Documentation on the refCount says:

Returns an observable sequence that stays connected to the source as long as there is at least one subscription to the observable sequence.

This is from v4 (can't find the same doc piece for v5, but that didn't change I guess).

In my example we have 0 subscriptions after 2 second and fromEvent is still connected to the source. So ref counting seems to be broken.

[cartant]

According to this PR, what you are seeing is the intended behaviour.

[ivan-kleshnin]

Ok, I'll give a few additional info. I'm making a multipage app. Each page has it's own set of listeners. Listeners from one page obviously should not interact with the listeners from the other page. Which is managed by subscription / unsubscription.

Now unsubscription doesn't work so I'll keep receiving events from "inactive" page which is weird and broken. If replayed sources does not take refCount into account (as they should, according to your own docs) there, at least, should be another method like destroy or something, that removes those event listeners.

I think it's pretty clear why unremovable event listeners are inappropriate. Memory leaks and repeating events are just some of the problems they bring.

When the refCount hits zero, and the source has neither completed nor errored

fromEvent source bound as event listener to the DOM node will never be completed or errored. So sticky, unremovable event listener is a real thing.

So I still think of it as a bug. If that is considered a desired behavior – please tell me some workarounds.

---

My first example was made up as a "simplest non-working case". Most of the time you share and not shareReplay DOM-sourced observables. But shareReplay freezing behavior influences them from the stateful stream down the code. "Spooky action at a distance" at it's best.

let intents = {
  click: fromDOMEvent(...)
}

// 100 lines below

let state = O.merge(intents.click ...)
  .scan(...)
  .shareReplay(1) // makes intents.click addEventListener unremovable

[cartant]

Given that the observable created with fromEvent is not going to complete, you can get the behaviour you want using multicast directly by replacing shareReplay(1) with multicast(() => new ReplaySubject(1)).refCount().

The implementation that preceded the referenced PR, was essentially the same, but included some special handling for sources that completed - which is not relevant here.

As far as I am aware, there is no documentation for shareReplay in RxJS version 5.

[ivan-kleshnin]

https://github.com/ReactiveX/rxjs/blob/master/MIGRATION.md says

shareReplay should be replaced with publishReplay().refCount()

which works (as well as your snippet above).

So, on the one hand, it's my fault of missing this doc piece, on the other – I can't get if shareReplay is deprecated or not (given that recent PR, the data is contradictory). Maybe you guys plan to fine-tune some new meaning for it – no problem, it's just would be great to leave a searchable note somewhere. shareReplay was a very useful operator in RxJS 4, so a proper explanation on "what's going on with it" should help many people.

[cartant]

shareReplay was added to RxJS in 5.4; it's not deprecated. The documentation could certainly be improved and there is a renewed effort to do so, but at the moment, some portions of it are very much a work in progress.

[ivan-kleshnin]

shareReplay was added to RxJS in 5.4;

Then why migration guide RX4 -> RX5 mentions it? And I certanly used it previously:

https://github.com/ivan-kleshnin/cyclejs-examples/blob/master/package.json#L32
https://github.com/ivan-kleshnin/cyclejs-examples/blob/7ec6ea0f7fc5aa7c3dd9f3e4b86a1d1ab75621c3/rx-utils.js#L10 (2016, Rx4)

Perhaps it was re-added to Rx5 in that PR.

Anyway, thanks for helping me!

[fjozsef]

I've run into this problem.
@benlesh Could you please confirm that this behavior is intended?

In my opinion it is very strange that the subscription (to the source) can leak even if there is no subscriber (on the shareReplay) anymore.

If the already given examples aren't enough then I can also add my use case where this kind of behavior is present as a bug.

[Jrubzjeknf]

@fjozsef Apparently it is intended per this bugfix.

I have also run into this problem. In our application we use the shareReplay() operator to prevent re-evaluation of the source observable (an expensive http-get) which has multiple subscribers. The change in behavior between RX4 and RX5 now causes a ObjectUnsubscribedError in multiple parts of our application.

Please reinstate the part in the migration guide where it is advised that shareReplay is replaced with pusblishReplay().refCount(), or at least make a specific note of it somewhere. This is definitely a breaking change, even if it was an unintended side-effect before.

[shlomiassaf]

Just use a custom operator, it's dead simple

export const weakShareReplay = <T>(bufferSize?: number, windowTime?: number ) =>
  (source: Observable<T>) =>
    source.pipe(publishReplay(bufferSize, windowTime), refCount());

@benlesh this could be added to the library, maybe?

[lock[bot]]

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.