RxJS: Share() an Observable & Emit Last Value to each new subscriber

Question

I'm using .share() to share an Observable among all subscribers to a service:

@Injectable()
export class ChannelsService {
    private store: IChannel[] = [];
    private _own: BehaviorSubject<IChannel[]> = new BehaviorSubject([]);
    readonly own: Observable<IChannel[]> = this._own.asObservable().share();
    ...(the rest of the service basically makes CRUD http request and then calls this._own.next(result) to emit the result to all subscribers)
}

Problem: Only the first subscription to the Observable (ChannelsService.own.subscribe(...)) is getting initial data, the rest of subscriptions get 'null' as a first subscription's value. Next calls to this._own.next(result) will emit its values to all subscribers properly.

Any idea of how to .share() an Observable among multiple subscribers and get the last value emitted for all the subscribers? (I've tried .share().last() but no way...).

Thanks!

score 5 · Accepted Answer · edited Sep 23 '17 at 16:56

5

You probably want shareReplay() (since RxJS 5.4.0) or publishReplay().refCount() in RxJS < 5.4.0.

For example:

this._own.asObservable()
  .publishReplay(1)
  .refCount()
  .take(1);

edited Sep 23 '17 at 16:56

Graham

7,431
18
59
84

answered Jun 30 '17 at 10:00

martin

93,354
25
191
226

Thank you for your answer and the detailed link, it worked! I would also recommend the reading of this article: https://blog.thoughtram.io/angular/2016/06/16/cold-vs-hot-observables.html#card_1501173301527_7527 It helped me to understand the underlying question: the difference between cold and hot observables. It recommend publishLast().refCount() that, from what I've seen, works the same way: it shares the same and unique subscription between all the subscribers and emit the last value on subscription. – aleixsuau Aug 16 '17 at 20:23

Jota.Toledo · Answer 2 · 2017-06-30T10:08:04.263

1

Some small improvement:

@Injectable()
export class ChannelsService {
    private store: IChannel[] = [];
    private _own: BehaviorSubject<IChannel[]> = new BehaviorSubject([]);
    get own(){
      this._own.asObservable();
    }
}

Now you can do the following in your components:

channels$: Observable<IChannel[]> = this.service.own;

And dont forget to unsubscribe from it if u manually subscribe to it

edited Jun 30 '17 at 10:08

answered Jun 30 '17 at 09:41

Jota.Toledo

27,293
11
59
73

Thanks for your comment @Jota.Toledo. Maybe your option is more elegant, but this way I'm able to call this.channelsService.own from components too... – aleixsuau Jun 30 '17 at 10:00
removing the share operator has no effect? – Jota.Toledo Jun 30 '17 at 10:03

Jorge Rodríguez Galán · Answer 3 · 2017-06-30T11:39:54.263

Your results are the expected using share without initialization value. Explanation

As commented before.. you have shortcuts to help you with this use case. Just a simple addition: giving "null" as initial value and filtering by those who are not.

const subject = new Rx.Subject();

const behaviorWithoutShortcut = subject
  .multicast(new Rx.BehaviorSubject(null))
  .refCount()
  .filter(function (x) { return x !== null; });

const behaviorWithShortcut = subject
  .publishBehavior(null)
  .refCount()
  .filter(function (x) { return x !== null; });

I just made an example jsbin example

RxJS: Share() an Observable & Emit Last Value to each new subscriber

3 Answers3