angular - RxJS:手动发出 Observable 的正确方法

Question

在 Angular 4.x 中使用 RxJS，我看到了两种截然不同的模式，用于从用户发起的操作流中生成 Observable。一个流是用户单击生成新对象的“添加项目”按钮的直接结果。另一个是我正在使用的一些第三方代码发出的一系列事件。

我希望能够使用“combineLatest”之类的东西组合这两个流，以生成单个 Observable。

我的按钮遵循以下模式:

const signal = Observable.create(
            (observer) => {
                this.additem= (item) => observer.next(item);
            }
        );

this.item$ = signal.map((item) => [item])
                            .scan((accumulator, value) => {
                                return accumulator.concat(value);
                            });

但是，我看到很多信息说我应该改用 Subjects - 我正在尝试将其与我的事件回调一起使用，如下所示:

sort$ = new Subject();

sortChange(sort){
        sort$.next(sort);
}

然后我尝试像这样组合这些:

combine$ = Observable.combineLatest(sort$, item$,
                  (sort, items) => {
                      return "something that does stuff with these";}
        );

我的问题是 - “手动”生成流的首选模式是什么？是否可以/应该像我在这里尝试做的那样将可观察对象和主题网格化为一个单独的可观察对象？

Answer 1

当然，您可以将 Observables 和 Subjects 合并为一个流。

我认为这里的问题是在您的用例中什么更有意义。根据您在实现类似“添加项目”功能时的描述，我更喜欢 Subject 而不是 Observable.create。

这是因为每次您订阅 signal 时，您都在重新分配 this.additem。为每个观察者调用 Observable.create 的回调。请注意，Observable.create 的更正确用法如下所示:

const signal = Observable.create((observer) => {
   this.additem = (item) => observer.next(item);
   return () => this.additem = null;
});

返回的回调 () => this.additem = null 在你取消订阅这个 Observable 时被调用，这是你应该处理所有清理的地方。

但是，如果您对 signal 进行两次订阅，那么您将覆盖 this.additem 两次，然后如果您选择取消订阅其中一个观察者，您将 this.additem = null 并且它可能会导致意外行为。

所以在这种情况下，使用 Subject 更有意义。例如像这样:

const subject = new Subject();
this.additem = (item) => subject.next(item);

如果你想看到更多关于 Observable.create 的真实例子，看看这个例子:Subscribe to a stream with RxJS and twitter-stream-api module

编辑:另请参阅 RxJS 5 首席开发人员的这些文章:

• https://medium.com/@benlesh/on-the-subject-of-subjects-in-rxjs-2b08b7198b93

• https://medium.com/@benlesh/rxjs-dont-unsubscribe-6753ed4fda87

• https://medium.com/@benlesh/hot-vs-cold-observables-f8094ed53339