Merge two observables, single output

Question

Hello guys I'm trying to grasp RxJS lib and the whole idea of reactive programming. I'm trying to merge two observables into one. First observable contains an array of objects DefectImages[] the second observable contains an array of strings, which then I convert to an array of DefectImages[]. After that I would like to merge these two observables into one.

Below my code:

const observable = CachedPhotosBuffer.getInstance().asObservable()
      .pipe(
        switchMap(data => {
          return data.map((url) => DefectImage.createTempImage(url, 'you', Date.now()));
        })
        );
    this.observable = Observable.create(observer => observer.next(this.defectImages));
    this.observable.pipe(
      merge(observable)
    ).subscribe(data => console.log('merge', data))

This KIND OF works as I expect BUT this merged observables Is connected to HTML Angular template.

<ion-list>
    **<ng-container *ngFor="let image of observable | async">**
      <ion-item *ngIf="image.deletedAt === undefined">
        <span class="item-container" (click)="showImage(image)">
          <ion-thumbnail item-start>
            <img id="{{image.url}}" src="{{getUrl(image) + image.url}}">
          </ion-thumbnail>
          <span>
            <p>created at: {{image.createdAt | date: 'd/M/yy H:m'}}</p>
            <p>created by: {{image.createdBy}}</p>
          </span>
        </span>
        <button ion-button item-end (click)="removeImage(image)">
          <ion-icon name="trash"></ion-icon>
        </button>
      </ion-item>
    </ng-container>
  </ion-list>

Console logs that I'm getting:

My question is Why do I have two separate logs for each stream instead of One console log with all the data combined?

Solution

Merging observables means that items emitted by both observable will be emitted successively and separately by the new merged observable, cf this page. If your observables emit just one item each and you want the merge the items by concatenating the arrays, you could use the zip operator as follows:

zip(observable, this.observable)
  .pipe(map(x => x[0].concat(x[1])))
  .subscribe(data => console.log('merge', data))

More precisely zip(obsa, obsb) creates a new observable that listens to obsa and obsb, and after receiving itema from obsa and itemb from obsb emits the item x=[itema, itemb]. In your case x[0]=itema, x[1]=itemb are arrays and (x => x[0].concat(x[1])) concatenates these two arrays. Note that if obsa and obsb emit more than one array, the zipped observable will always wait to have one item from obsa and one from obsb before emitting a new [itema, itemb]. For the concat() method, cf this page.

And don't forget to import { zip } from 'rxjs' and import { map } from 'rxjs/operators'.

Comments

[filemonczyk]

works like a charm, Could you explain why we need map operator? what we have before map and after map ? and why is this implemented in such way x[0].concat(x[1])??

[Max]

Instead of x[o].concat(x[1]) you could use x.flat(). It is more concise and already widely supported: caniuse.com/array-flat