是否可以通过 Rx 观察到任何计数器？

Question

我正在寻找我会调用CounterObservable 的东西，一侧将计算其上的数字，另一侧将是观察者一侧，每次总计数发生变化时都会收到通知。

换句话说，我会有这样的东西

public CounterObservable totalMailsReceived = new CounterObservable(0);

public void OnNewMail(Mail mail) 
{
   totalMailsReceived++;
   ///Rest of the code goes here
}

在观察者方面，我将拥有

mailManager.totalMailsReceived.Subscribe(count => labelCount.Text = count.ToString());

或者，如果我想要真正优雅，我会使用 Paul Betts 的 ReactiveUI，如下所示

mailManager.totalMailsReceived.ToProperty(x => x.TotalMailsReceived);

到目前为止，我在 Rx 中没有发现任何可以帮助我的东西。但我想如果我创建自己的类来实现IObservable<int>。我正在考虑使用Sample MSDN Code for IObservable implementation。

我的问题是 1. 那是MSDN Sample 线程安全的吗？ 2. Rx 中真的没有任何东西可以做我想做的事情吗？

Answer 1

只需使用BehaviorSubject：

public class MailServer
{
    private BehaviorSubject<int> _count = new BehaviorSubject<int>(0);

    public IObservable<int> TotalMailsReceived
    {
        get { return _count; }
    }

    public void OnNewMail(Mail mail)
    {
        _count.OnNext(_count.Value + 1);
    }
}

或者，如果您决定更深入地研究 Rx，以便您只是观察 Mail 流，那么您可以使用 Scan 运算符，这对那个有好处，而 Publish 记住最近的值和将其多播给所有订阅者。

你可以编写这个新的扩展方法：

public IObservable<T> RunningTotal<T>(this IObservable<T> source)
{
    return source.Scan(0, sum => sum + 1);
}

然后像这样使用它：

public class MailServer
{
    private IConnectableObservable<int> _total;
    private IDisposable _subscription;

    public MailServer(IObservable<Mail> incomingMail)
    {
        _total = incomingMail.RunningTotal().Publish(0);
        _subscription = _total.Connect();
    }

    public IObservable<int> TotalMailsReceived
    {
        get { return _total; }
    }
}