如何使用反应式扩展展开可观察对象

Question

我是一名有用的助手，可以为您翻译文本。

我希望你能够将这个服务转换成Rx Observable，该服务具有以下接口：

    IEnumerable<Price> FetchUpdatedPrices()
    {
       //do work to return changed data since last update
    }

我的想法是使用rx让消费者订阅更新。实现方式是每隔x秒轮询服务并调用观察者。

我想到了以下解决方案。

    public IDisposable Subscribe(IObserver<IEnumerable<Price>> observer)
    {

        IObservable<IEnumerable<Price>> updatedPrices = Observable.Interval(new TimeSpan(0, 0, 1))
            .Select(r => FetchUpdatedPrices());

        return updatedPrices.Subscribe(observer);
    }

问题是我希望观察者看到一个IObservable<Price>而不是IObservable<IEnumerable<Price>>。有没有人能给这个Rx新手一些指针如何做到这一点？

- Fen

Answer 1

那么 SelectMany 怎么样呢？

IObservable<IEnumerable<Price>> updatedPrices = Observable.Interval(new TimeSpan(0, 0, 1))
    .SelectMany(r => FetchUpdatedPrices());

- Kana Natsuno

---

很简单，谢谢。我猜当时我看这个的时候可能有点累了 :s - Fen

这句话不应该是 IObservable<Price> updatedPrices = Observable.Interval(new TimeSpan(0, 0, 1)).SelectMany(r => FetchUpdatedPrices().ToObservable()); 吗？ - Enigmativity

加入一个组播（Multicast）吧，否则每个订阅者将会拥有自己的计时器。 - Ana Betts