为 RX Stream / IObservable 添加一个初始值

Question

我有一个方法可以将过滤后的 RX 流作为 Iobservable 返回：

public IObservable<Price> LivePrices(Instrument instrumentDetails)
        {
            return _priceObserver.Stream
                .Where(o => o.Symbol == instrumentDetails.Symbol )
                .Select(o => GetPrice(o, instrumentDetails));
        }

问题是对于某些值，流不会经常更改，因此我需要使用第一个值对其进行初始化

我该怎么做？我读到Subject 既可以是观察者，也可以是可观察者。所以我想我需要以某种方式将其订阅为Subject，在流中添加第一条消息，然后将其设置为现在的样子。但不知道该怎么做

有什么想法吗？

Answer 1

你试过StartWith吗？

public IObservable<FxDeal> LiveRates(NegotiationDetails negotiation)
{   
    var obs = _ratesObserver.Stream
        .Where(o => o.BaseCurrency == negotiation.Base && o.TermsCurrency == negotiation.Terms)
        .Select(o => GetFxDeal(o, negotiation));

    return condition ? obs.StartWith(new FxDeal()) : obs;
}