как правильно составить наблюдаемое

Question

Для начала в моем компоненте я объявляю наблюдаемое ...

import { Observable } from 'rxjs/Observable';
import { Subscription } from 'rxjs/Subscription';
import { tap, map, concat } from 'rxjs/operators';

contactsDataSource$: Observable<any>;

затем я назначаю поток наблюдаемой из службы http:

this.contactsDataSource$ = this.medialistsrv.getMediaListBuilderData({
    mediaListId: this.mediaListId,
    sortType: this.sortType,
    startRow: this.startRow,
    numRows: this.numRows
}).pipe(
    tap(response => {
        this.mediaList = response
            .filter(items => items.inMediaList === true)
            .map(items => items.contactId);
        this.mediaList.forEach(id => {
            this._contacts.push(this.fb.control(id));
        });
    }),
    map(response => response)
);

и введите его в шаблон с помощью асинхронного канала:

<div *ngIf="(contactsDataSource$ | async) as contactList; else loading">

позже в шаблоне я ввожу метод для загрузки большего количества результатов. это где мне нужно, чтобы конкат к начальному потоку

increment() {
    this.contactsDataSource$.concat(
        this.medialistsrv.getMediaListBuilderData({
            mediaListId: this.mediaListId,
            sortType: this.sortType,
            startRow: (this.numRows + 1),
            numRows: (this.numRows * 2)
        }).map(response => response)
    );
}

при запуске я получаю сообщение об ошибке в консоли:

ng:///MediaListModule/MediaListBuilderComponent.ngfactory.js:351 ERROR TypeError: this.contactsDataSource$.concat is not a function

Я все еще на учебной скале rxjs, поэтому, очевидно, я не понимаю, как сделать это правильно. Что мне здесь не хватает?

Answer 1

Проблема с тем, как вы это делаете, заключается в том, что подписка будет завершена после первого звонка на getMediaListBuilderData. Вам необходимо настроить наблюдаемое и то, как оно объединяет результаты из нескольких источников перед первой подпиской. Кроме того, если вы хотите, чтобы действия в канале происходили с каждым результатом, который нужно поместить после объединения потоков.

Создайте наблюдаемое из concat, где вы передадите свой первоначальный вызов, и Subject, которое будет передано методам подписки всех вызовов getMediaListBuilderData в increment().

this.incrementsSubject = new Subject<myType>();
this.contactsDataSource$ = 
    concat(
        this.medialistsrv.getMediaListBuilderData({
            mediaListId: this.mediaListId,
           sortType: this.sortType,
           startRow: this.startRow,
           numRows: this.numRows
        }
        , this.incrementsSubject)
    .pipe(
        tap(response => {
            this.mediaList = response
               .filter(items => items.inMediaList === true)
               .map(items => items.contactId);
            this.mediaList.forEach(id => this._contacts.push(this.fb.control(id)));
        }),
    );

increment() {

    this.medialistsrv
        .getMediaListBuilderData({
            mediaListId: this.mediaListId,
            sortType: this.sortType,
            startRow: (this.numRows + 1),
           numRows: (this.numRows * 2)
         })
         .subscribe(this.incrementsSubject);
}