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

Question

Let's say I would like to get a set of records from the repository, display them in a list using *ngFor , for example.

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

Now the user clicks the "Create ..." button, another record is added to the repository and

 recordAdded: EventEmitter<string>; 

lights up to tell me the location. So I get this new entry - and only this entry - from the store and ... yells, how do I get my *ngFor to display this extra entry?

Ok, so I could store all the records in an array, like

 _records: Record[]; 

and populate this array by subscribing to Observable<Record[]> as

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

But this array must be Observable itself in order to notify consumers of the appearance of 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 it painful to watch, there is also a ton of overhead, since the entire array is republished each time a new record is added and, most likely, Angular 2 will rebuild the entire list from scratch at every step.

Since this is such a common scenario, I believe there should be a much better way to do this.

Answer 1

Yes, of course - you need to use flatmap

flatmap is a way to go when you sign up for a subscription.

You have a stream of threads (meta stream), and you need to smooth it out - then you can subscribe once to a flattened stream (avoiding a nested subscription).

read the tutorial until the bit in which it is presented to you is raster.

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

Here is some code to beat you back ...

 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(); 

Do not forget .... publish (). refCount ()

or you will get some requests, etc.

read more about this on the link I posted.

PS

In the code above, q is the value emitted from the initial observable. Keep in mind that observation conditions and flow are interchangeable; this is one and the same. observable simply describes the stream you can subscribe to. Consequently, reactive programming (Rx). You are responding to events originating from a stream.