完成的冷可观察集合

Question

所以也许我把这个观察者模式弄错了，但这就是我想要的。我有一种方法应该从数据库中获取一些整数。看起来像这样：

IObservable<int> GetInts()

每次他得到一个整数时，他自然应该通过 OnNext() 告诉它的订阅者。但是，如果观察员迟到，他可能会错过其中一些电话。我仍然希望他知道所有信息，所以当他订阅 Observable 时应该“填写”并告诉他他错过的信息......然后在添加新内容时继续告诉他。

但是，在某些时候，数据库读取将结束，并且该方法会知道这一点。此时，他应该调用订阅者的 OnComplete()。所以基本上我想要的是一个向人们讲述一些东西的集合。我还没有找到可以做这样的事情的东西，所以我必须实现它。它可能看起来像：

class MyObservable:List<int>,IObservable<int>{
private bool completed;
private List<IObserver<int>> observers = new List<IObserver<int>>();

public bool Completed{
    get{return completed;}
    set{
        completed = value;
        if(completed == true){
            observers.ForEach(obs=>obs.OnCompleted());
        }
    }
}

public IDisposable Subscribe(IObserver<int> obs){
    observers.Add(obs);
    this.ForEach(n => obs.OnNext(n));
    if(Completed) obs.OnCompleted();
    return Disposable.Create(()=>observers.Remove(obs));
}

public void Add(int n){
    observers.ForEach(obs => obs.OnNext(n));
    base.Add(n);
}   
}

有没有更好的方法来做到这一点？像这样的东西已经在 Rx 中实现了？或者，如果我根本不应该这样做，我最好的选择是什么？

如果这是要走的路，我应该担心什么样的事情？ （例如并发..等）。

Answer 1

你想要一个ReplaySubject<int>。它会记住它曾经收到的所有更新。当观察者订阅它时，观察者会收到主题的历史记录，然后是所有新的更新。

试试这个例子：

var subject = new ReplaySubject<int>();
subject.Subscribe(n => Console.WriteLine("A: {0}", n));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
Console.ReadLine();
subject.Subscribe(n => Console.WriteLine("B: {0}", n));
Console.ReadLine();
subject.OnNext(4);
subject.OnNext(5);

您应该得到如下交互：

A: 1
A: 2
A: 3
<Press Enter>
B: 1
B: 2
B: 3
<Press Enter>
A: 4
B: 4
A: 5
B: 5