使用RxJS 5可观察量的延迟模式

Question

对于任意承诺实现,延迟模式(不要与反模式混淆)可能看起来像:

const deferred = new Deferred;
...
// scopes where `deferred` object reference was passed before promise settlement
deferred.promise.then((result) => { ... }, (error) => { ... });
...
deferred.resolve(...);
// doesn't affect promise state
deferred.reject();
...
// after promise settlement
deferred.promise.then((result) => { ... }, (error) => { ... });

deferredobject保存未定义的promise,可以通过引用传递给其他函数作用域.所有承诺链都将在承诺结算时执行,无论是否deferred.promise在链接之前then或之后结算都无关紧要.承诺状态在结算后无法改变.

---

正如答案所暗示的那样,最初的选择是ReplaySubject和AsyncSubject.

对于给定的设置(演示)

var subject = new Rx.AsyncSubject;
var deferred = subject.first();

deferred.subscribe(
  console.log.bind(console, 'Early result'),
  console.log.bind(console, 'Early error')
);

setTimeout(() => {
  deferred.subscribe(
    console.log.bind(console, 'Late result'),
    console.log.bind(console, 'Late error')
  );
});

这导致了理想的行为:

subject.error('one');
subject.next('two');

早期错误一

迟到的错误一

这会导致不良行为:

subject.error('one');
subject.next('two');
subject.complete();

早期错误一

后期结果二

这会导致不良行为:

subject.next('two');
subject.complete();
subject.next('three');

早期结果二

后期结果三

结果ReplaySubject不同但仍与预期结果不一致.next值和error错误是分开处理的,并complete不会阻止观察者接收新数据.这可能适用于单个next/ error,问题是next或者error可能无意中多次调用.

使用的原因first()是因为subscribes是一次性订阅,我想删除它们以避免泄漏.

如何使用RxJS observable实现它？

Answer 1

您可能正在寻找一个Rx.ReplaySubject(1)（或一个Rx.AsyncSubject()取决于您的用例）。

有关主题的更详细说明，请参阅不同 RxJS 主题的语义是什么？。

基本上，主题可以通过引用传递，就像延迟一样。只要您持有该引用，您就可以向其发出值（解析将是'next'(Rxjs v5) 或'onNext'(Rxjs v4) 后跟'complete'或）。'onCompleted()'

您可以拥有任意数量的主题订阅者，类似于then延期订阅者。如果您使用 a replaySubject(1)，任何订阅者都将收到最后发出的值，该值应该回答您的it doesn't matter if deferred.promise was settled before chaining with then or after.. 在 Rxjs v4 中，areplaySubject将在完成后将其最后一个值发送给订阅者。我不确定 Rxjs v5 中的行为。

• https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/asyncsubject.md
• https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/replaysubject.md

更新

使用 Rxjs v4 执行以下代码：

var subject = new Rx.AsyncSubject();
var deferred = subject;

deferred.subscribe(
  console.log.bind(console, 'First result'),
  console.log.bind(console, 'First error')
);

setTimeout(() => {
  deferred.subscribe(
    console.log.bind(console, 'Second result'),
    console.log.bind(console, 'Second error')
  );
});

subject.onNext('one');
subject.onCompleted();
subject.onNext('two');
subject.onNext('three');
subject.onNext('four');

产生以下输出：

First result one
Second result one

然而，使用 Rxjs v5 执行的相同代码却不会：

First result one
Second result four

所以基本上这意味着Rxjs v5 中主题的语义已经改变！这确实是一个需要注意的重大变化。无论如何，您可以考虑回到 Rxjs v4，或者使用 artur grzesiak 在他的答案中建议的周转方案。您还可以在 github 站点上提交问题。我认为更改是有意为之，但实际上并非如此，提交问题可能有助于澄清情况。无论如何，无论选择什么行为，都应该正确记录。

关于主题语义的问题有一个链接，显示与多个和延迟订阅相关的异步主题