如何强制 publishReplay() 重新订阅？

Question

我对 RxJS 很陌生，这是我的故事。我想要一个可观察的“会话”，因此新订阅者将始终获得当前会话，然后是所有新会话（如果它们可以出现）。所以我写了这样的东西：

var session = Rx.Observable.from([0,1,3])
  .do(x => console.log("Useful job"))
  .publishReplay(1)
  .refCount();

var subscr1 = session.subscribe( x => {
    console.log("sub1 = " +x)
    //subscr1.unsubscribe();
})

console.log("Completed");
subscr1.unsubscribe();

session.subscribe( x => {
  console.log("sub2 = " +x)
});

输出是：

Useful job
sub1 = 0
Useful job
sub1 = 1
Useful job
sub1 = 3
Completed
sub2 = 3

为什么 sub2 订阅时没有有用的工作？我预计它会完全冷！

Answer 1

.publishReplay() 的第一个参数是它将重播的项目数，因此如果您知道始终只能期望 3 个，则可以使用 .publishReplay(3)。

如果您想重播整个序列，您可以先使用 toArray() 收集其所有项目，将数组存储在 ReplaySubject 中，然后将数组展平为单个值。

var session = Rx.Observable.from([0,1,3])
  .toArray()
  .publishReplay(1)
  .refCount()
  .concatAll();

请注意，toArray() 在其源代码完成之前不会发出任何内容。

编辑：我知道现在有什么问题了。

您的方法是正确的，但问题出在Observable.from 和publishReplay() 中使用的主题。源 Observable 发出三个项目并发送complete 通知。当主题收到complete 或error 通知时，它会将自己标记为stopped，并且不会再重新发送任何项目。这正是您的示例中发生的情况。

如果您手动发出值而不发送complete，它将按预期工作。

// var session = Rx.Observable.from([0,1,3])
var session = Rx.Observable.create(subscriber => {
    subscriber.next(0);
    subscriber.next(1);
    subscriber.next(3);
  })
  .do(x => console.log("Useful job"))
  .publishReplay(1)
  .refCount();

这会打印到控制台：

Useful job
sub1 = 0
Useful job
sub1 = 1
Useful job
sub1 = 3
Completed
sub2 = 3
Useful job
sub2 = 0
Useful job
sub2 = 1
Useful job
sub2 = 3

看到同样的问题：Rx.Subject loses events

Answer 2

我终于找到了解决办法。

将 .publishReplay(1) 替换为 .multicast(() => new Rx.ReplaySubject(1))。所以基本上我用subjectFactory替换了一个主题。

然后它按预期工作。所以当所有人都退订时，它变得绝对冷。