是否可以重新启动 IConnectableObservable?

Question

我正在尝试弄清楚如何在IConnectableObservable完成或出错后"重新启动"它。

下面的代码显示了可连续可观察对象的两个订阅者(A、B)。释放订阅后，将有新的订阅者 (C) 连接。我希望它出现在 C 看来它是一个全新的可观察对象，但我只在第一次订阅中得到异常。

static void Main(string[] args)
{
var o = Observable.Create<string>(observer =>
{
observer.OnNext("msg");
observer.OnError(new Exception("boom"));
return Disposable.Create(() => {
Console.WriteLine("Observer has unsubscribed");
});
}
)
.Publish();
o.Subscribe(
x => Console.WriteLine("A: " + x),
ex => Console.WriteLine("A: " + ex),
() => Console.WriteLine("A: done"));
o.Subscribe(
x => Console.WriteLine("B: " + x),
ex => Console.WriteLine("B: " + ex),
() => Console.WriteLine("B: done"));            
var subscription = o.Connect();
subscription.Dispose();
o.Subscribe(
x => Console.WriteLine("C: " + x),
ex => Console.WriteLine("C: " + ex),
() => Console.WriteLine("C: done"));        
subscription = o.Connect();
}

给出以下结果：

A: msg
B: msg
A: System.Exception: boom
B: System.Exception: boom
Observer has unsubscribed
C: System.Exception: boom
Observer has unsubscribed

而我想要：

A: msg
B: msg
A: System.Exception: boom
B: System.Exception: boom
Observer has unsubscribed
C: msg
C: System.Exception: boom
Observer has unsubscribed

有什么想法吗？谢谢！

Answer 1

虽然它不会"重新启动"可观察量，但将Publish替换为Replay可提供您期望的输出。但是，请记住，这将缓冲来自可观察源的所有值。最好限制重播值的数量。