如何正确连接可观察对象

Question

从我的组件开始，我声明了 observable...

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

contactsDataSource$: Observable<any>;

然后我将流分配给来自 http 服务的 observable：

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 后完成。您需要在第一次订阅之前设置 observable 以及它如何连接来自多个来源的结果。此外，如果您希望管道中的操作与每个结果一起发生，则它需要放在流的串联之后。

从 concat 创建一个 observable，您将在其中传递您的初始调用和一个 Subject，该 Subject 将传递给 increment() 中所有 getMediaListBuilderData 调用的订阅方法。

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