c# - 你如何用响应式扩展扇出 observables

Question

我有一个服务，我想把它变成一个 rx observable

服务有一个接口(interface)

    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 菜鸟任何关于如何做到这一点的建议？

Answer 1

SelectMany怎么样？ ？

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