Angular 2 + RxJS: async pipe with .share() operator

Question

When using the async pipe on a observable that is using the .share() operator (due to expensive calculations in the backend), I stumbled upon this behaviour:

data$ = (new Observable(observer => {
    let counter=0;
    observer.next(counter)

    window.setInterval(() => {
      observer.next(counter);
      counter++;
    }, 2000);
  }))
  .share();

Template:

{{ (data$|async) !== null }}
{{ (data$|async) !== null }}

The output of the initial value is:

true false

The following outputs (after more than 2 seconds) are:

true true

which is the behavior I would expect for the first value, too. If I omit the .share(), the output for the first value is "true true", as I would expect. I suppose the behavior above is due to the fact that the first expression in the template triggers observable execution, and once the second async pipe subscribes to the observable, the data is already gone. Is this explanation correct? And how can I avoid this behavior?

Since there is supposed to be a single value, it is `observer.next(...); observer.complete()`. — Estus Flask, Sep 20 '16 at 20:16
the example with a single value was just for demonstration, in my app the observable has multiple values — highwaychile, Sep 20 '16 at 21:57
I edited my question with an example code: The first value of the observable triggers the strange behavior, while the later ones don't. Do you maybe know a way to make the first value available for both async pipes? — highwaychile, Sep 20 '16 at 22:23
I guess I do. In RxJS4 `shareReplay(1)` was used (as the name says, it shares the replay of last 1 value). For A2+RxJS5 a direct counterpart can be used, see the answer. — Estus Flask, Sep 20 '16 at 22:56

score 16 · Accepted Answer · edited May 23 '17 at 12:02

16

According to the reference,

The async pipe subscribes to an Observable or Promise and returns the latest value it has emitted.

In RxJS 4, shareReplay is used to achieve the desired behaviour.

In RxJS 5, a direct counterpart to shareReplay is publishReplay followed by refCount (see the explanation and the discussion).

So it should be

data$ = (new Observable(observer => { ... }))
.publishReplay(1)
.refCount();

edited May 23 '17 at 12:02

Community

1
1

answered Sep 20 '16 at 22:52

Estus Flask

206,104
70
425
565

But why did it show true and then false ? ( at time=0) – Royi Namir Mar 29 '18 at 14:42
Because at time 0 data$ contains one value (`0`), and after subscription with first `(data$|async)` it's 'exhausted'. Since there is no second value in data$, and `(data$|async)` expression should resolve to some value synchronously, it resolves to `null` (this is how `async` pipe works). – Estus Flask Mar 29 '18 at 14:52
Wait , at time=0 , how many subscribers exists ?1 or 2 ? ( i'm asking it becuase share sees refcounts. ). Can you please explain this in a refcount POV ? – Royi Namir Mar 29 '18 at 15:06
`{{ (data$|async) !== null }} {{ (data$|async) !== null }}` creates 2 subscriptions, and for shared observable there are 2 refcounts. – Estus Flask Mar 29 '18 at 15:25
estus , I know this but what happens for non-shared observable ? this is what i'm trying to understand - how come the first is getting true while the second one also subscribed , which means refcount=2 ... ? – Royi Namir Mar 29 '18 at 15:26
In non-shared observable, `observer => { ... }` is called every time data$ is subscribed, and each subscription gets a value that is produced with `observer.next(0)`. In shared observable, it's called once, and the value gets the first subscription. I'd suggest to stay away from async and true/false, because they make things more complicated. Try `data$.subscribe(console.log); data$.subscribe(console.log)` on shared observable and you'll see that only 1 log entry is produced, by first subscription. – Estus Flask Mar 29 '18 at 15:41
Thank you for your time and help. ^^ – Royi Namir Mar 29 '18 at 15:45
I've tried what you said and it's logged twice : http://jsbin.com/bayibopazi/edit?js,console – Royi Namir Mar 29 '18 at 15:48
Glad if it helped. That's because you used Observable.of, it produces completed observable. It will be different for `new Observable(o => o.next(1))`, http://jsbin.com/gexijovose/1/edit?js,console – Estus Flask Mar 29 '18 at 15:55
estos - I've asked a new question ( to not pollute this thread) : it seems that I'm still missing something : https://stackoverflow.com/questions/49561125/rxjss-share-operator-behaves-differently-when-complete – Royi Namir Mar 29 '18 at 16:38
@RoyiNamir I'll look into it, but it seems that you summed it up well. Observables are automatically unsubscribed on completion, that's the point. There are some RxJS guys on SO who can possibly explain this better than I did. – Estus Flask Mar 29 '18 at 16:55
1

What about RxJS 6? There is still (or again?) `shareReplay()`. And this blog post says it's also part of RxJS 5.4 ( https://blog.angularindepth.com/rxjs-whats-changed-with-sharereplay-65c098843e95 ). Or is this something different than version 4's `shareReplay()`? – Benjamin M Jun 18 '19 at 13:38
@BenjaminM Indeed, and the link you posted does a good job at explaining that. I suppose it should be used as [this example](https://gist.githubusercontent.com/cartant/1b45497bbb560ea2d9552ce2d3e3421d/raw/795df7476c51b85a22cf9951ef4e20b9d96f1f91/share-replay-interval-ref-count.ts) shows, `shareReplay({ bufferSize: 1, refCount: true })` because default options result in different behaviour. It doesn't seem more expressive than `.publishReplay(1).refCount()` to me. – Estus Flask Jun 18 '19 at 20:02

Angular 2 + RxJS: async pipe with .share() operator

1 Answers1

Linked