RxJS 过滤多个流

Question

其实我是被这个问题困住了。

我有一个使用 Angularfire2 的事件流。

this.events$ = this.af.database.list('/events')
    .map(events => {
    const filtered = events.filter(event => event.title === "Title 1");
    return filtered;
    });

这样检索我的数据：

{
  title: "Title 1",
  userid: "1"  
}

然后我有第二个这样的用户流：

this.users$ = this.af.database.list('/users/1')
    .map(users => {
    const filtered = users.filter(user => user.name === "Name 1");
        return filtered;
    });

现在我想合并流以获得以下结果：

将event.title === "Title 1" 和user.name === "Name 1" 的所有事件返回给我

这怎么可能？

提前致谢！

Answer 1

这个答案已经过时了！很遗憾，我无法删除它。

这样的东西对我有用：

this.events$ = this.af.database.list('/events')
      .map(events => {
        events.map(event => {
            this.af.database.object('users/' + event.userid)

            .subscribe(user => {
                event.user = user;
              })
            })
             
            const filtered = events.filter(event => event.title === "Test 1")
                .filter(event => event.user.name === "Name 1");;
                return filtered;
        });

我正在订阅我的 Event observable 中的用户。之后，我将用户保存到我的事件的属性中。

工作对我来说很好！

注意：如果您想获取用户的实时更新，请将 observable 保存到属性而不是用户对象。

Answer 2

如果你想合并两个 Observable，请使用 merge():

观看现场演示：http://plnkr.co/edit/7j71Qzok7CoFVt49QBs4

import {Observable, Subject} from 'rxjs';

let stream1 = new Subject();
let stream2 = new Subject();

Observable.merge(stream1, stream2)
  .subscribe(r => {
    console.log(r);
  });

stream1.next(43);
stream2.next(42);
stream1.next(41);

打印到控制台：

43
42
41

如果您想在所有 Observable 完成后发出单个值，请使用 forkJoin()：

import {Observable, Subject} from 'rxjs';

let stream1 = new Subject();
let stream2 = new Subject();

Observable.forkJoin(stream1, stream2)
  .subscribe(r => {
    console.log(r);
  });

stream1.next(43);
stream2.next(42);
stream1.next(41);

stream1.complete();
stream2.complete();

打印到控制台：

[41, 42]