RxJS Observable 合并角度

Question

我有一个订阅消息流的沙箱，我想过滤该流以查找使用另一个组件中指定的路由参数发送给特定用户或从特定用户接收的消息。

messages.sandbox.ts：

messages$: Observable<Array<Message>> = this.store.select(state => state.data.messages);    

fetchReceived(id: string): Observable<Array<Message>> {
    return this.messages$.map((messages: any) => {
        return messages.filter((message: Message) => {
            return message.recipientId == id;
        });

    });
}

fetchSent(id: string): Observable<Array<Message>> {
    return this.messages$.map((messages: any) => {
        return messages.filter((message: Message) => {
            return message.userId == id;
        })
    })
}

messages.detail.container.ts

sentMessages$ = new Observable<Array<Message>>();
receivedMessages$ = new Observable<Array<Message>>();
matchingMessages$ = new Observable<Array<Message>>();



ngOnInit() {

    this.route.params.subscribe((params: Params) => {
        this.sentMessages$ = this.sb.fetchReceived(params['id']);
        this.receivedMessages$ = this.sb.fetchSent(params['id']);

        this.matchingMessages$ = Observable.merge(this.sentMessages$, this.receivedMessages$);

    });
}

this.matchingMessages$ 似乎只包含 this.receivedMessages$ 但是我知道 this.sentMessages$ 不为空，因为我可以在模板中毫无问题地使用它。

我在合并 Observables 时遗漏了什么吗？创建一个单独的 fetchMessages 方法来过滤 userId 或 recipientId 等于路由参数 id 会更好吗？如果是这样，我会怎么做？

谢谢！！

Answer 1

您的总体思路是正确的。只是一些缺陷。

1. 切勿使用new Observable<T>()。它不会做你认为它会做的事情。它几乎没有做任何有用的事情。始终从工厂方法或其他可观察对象构造可观察对象
2. 您需要使用运算符将​​可观察参数转换为新的可观察对象。您的问题是您 订阅 可观察的参数，然后每次都构造新的可观察对象。但是其他代码已经订阅了初始的 observables，所以他们永远不会看到这些变化。

所以你想做这样的事情：

sentMessages$ : Observable<Array<Message>>;
receivedMessages$ : Observable<Array<Message>>;
matchingMessages$ : Observable<Array<Message>>;

ngOnInit() {
    const params$ = this.route.params;

    // use switchMap to map the new params to a new sent observable
    // each time params change, unsubscribe from the old fetch and subscribe
    // to the new fetch.  Anyone subscribed to "sentMessages" will see the
    // change transparently
    this.sentMessages$ = params$.switchMap((params: Params) => this.sb.fetchReceived(params['id']));
    // same for received
    this.receivedMessages$ = params$.switchMap((params: Params) => this.sb.fetchSent(params['id'])));

    // merge the 2 streams together
    this.matchingMessages$ = Observable.merge(this.sentMessages$, this.receivedMessages$);
}

编辑：

回答您的另一个问题：创建一个匹配发送者和接收者的单个可观察对象是否更好：取决于您的用例。但你可以这样做：

messages.sandbox.ts：

fetchEither(id: string): Observable<Array<Message>> {
    return this.messages$.map((messages: any) => {
        return messages.filter((message: Message) => {
            return message.recipientId == id || message.userId === id;
        });

    });
}

容器：

matchingMessages$ : Observable<Array<Message>>;

ngOnInit() {
    const params$ = this.route.params;

    // use switchMap to map the new params to a new either observable
    // each time params change, unsubscribe from the old and subscribe
    // to the new fetch.  Anyone subscribed to "matchingMessages" will see the
    // change transparently
    this.matchingMessages$ = params$.switchMap((params: Params) => this.sb.fetchEither(params['id']));
}