How to build a true observable collection using RxJS and Angular2?

Question

Let's say I'd like to retrieve a set of records from a store, display them in a list using *ngFor, e.g.

<ul>
    <li *ngFor="let record in records | async">
        ...
    </li>
</ul>

Now the user clicks the 'New...' button, another record is added to the store and

recordAdded: EventEmitter<string>;

fires to tell me about the location. So I get that new record - and that record only - from the store and... whoops, how do I get my *ngFor to display this additional record?

Ok, so I could keep all records in array, for example

_records: Record[];

and fill this array by subscribing to the Observable<Record[]> like

this.recordService.getAll().subscribe(r => this._records = r);

But this array needs to be an Observable itself in order to notify consumers when there is a new record. So

observableRecords = Observable.create(obs => {
    this.recordService.getAll().subscribe(rec => {
        this._records = rec;
        obs.next(rec);
        // Got the array published, now waiting for events
        this.recordAdded.subscribe(rec => {
            this._records.push(rec);
            obs.next(this._records);
        });            
    });
});

Ugh... Not only is this excruciating to look at, there's also a ton of overhead as the whole array gets re-published every time a new record is added and - most likely - Angular 2 will re-build the whole list from scratch at every turn.

Since this is such a common scenario I imagine there must be a much better way to do this.

Answer 1

yeah, there certainly is - you need to use flatmap

flatmap is the way to go whenever you find yourself subscribing within a subscribe.

You've got a stream of streams (a metastream) and you need to flatten it - then you will be able to subscribe once to the flattened stream (avoiding the nested subscribe).

read the tutorial until the bit where it introduces you to flatmap.

https://gist.github.com/staltz/868e7e9bc2a7b8c1f754

Here's some code to kick you off ...

public tagsTextStream = this.tagsTextSubject.asObservable().flatMap((q:string) => {
// noinspection UnnecessaryLocalVariableJS
let restStream = this.restQueryService.getTagsList(q)
  .map((tags:any) => {
    // filter out any tags that already exist on the document
    let allTags = _.map(tags, 'name');
    let documentTags = _.map(this.tags, 'name');
    return _.pull(allTags, ...documentTags);
  })
  .catch((err:any) => {
    return Observable.throw(err);
  });
return restStream;
}).publish().refCount();

Don't forget ... .publish().refCount()

or you end up with multiple requests and stuff.

read more about it all on the link i sent.

PS

In the code above, q is the values being emitted from the initial observable. Be advised that the terms observable and stream are interchangeable, they are the same thing. observable is just describing a stream you can subscribe to. Hence, reactive programming (Rx). You are reacting to events coming from the stream.