异步方法中的 Rx StartWith 不应用起始值 [英] Rx StartWith in async method not applying the starting value

问题描述

我正在尝试使用 StartWith 方法将初始值注入 RX 流:

I am trying to inject an initial value into an RX stream using the StartWith method:

public async Task<IObservable<Price>) Stream(Instrument instrumentDetails)
{
    var initialPrice = await _svc.GetSomeInitialPrice();

    var stream = _priceObserver.Stream
        .Where(o => o.Symbol == instrumentDetails.Symbol)
        .Select(o => GetPrice(o, instrumentDetails));

    stream.StartWith(initialPrice);

    return stream;
}

然而，由于调用获取初始值，该方法是异步的.并且无论如何它都需要在这个调用堆栈中一直保持异步

however, the method is async due to the call to get the initial value. and it needs to be async all the way through this call stack anyway

我发现该值从未添加到开头.我只得到流的其余部分

I am finding the value is never added to the start. I just get the rest of the stream

如果我 await StartWith 方法，它永远不会返回

if I await the StartWith method, it never returns

任何想法我做错了什么

推荐答案

IObservable 上的方法不会修改底层对象——它们返回一个新对象.stream.StartWith(initialPrice) 返回一个你忽略的新 observable，它对 stream 没有任何作用.

The methods on the IObservable don't modify the underlying object - they return a new one. stream.StartWith(initialPrice) returns a new observable which you ignore, it doesn't do anything to stream.

你应该这样写:

stream = stream.StartWith(initialPrice);

或者:

var stream = _priceObserver.Stream
    .Where(o => o.Symbol == instrumentDetails.Symbol)
    .Select(o => GetPrice(o, instrumentDetails))
    .StartWith(initialPrice);

旁注:如果您等待一个 observable，它将一直等到 observable 完成，即当它发出所有值并调用其 OnComplete 方法时.您通常应该等待一个 observable，您知道它只会发出 1 个值然后完成(例如，对远程服务器的请求)，因为等待它只会返回最后发出的值.因此，如果您的 stream 预计会持续发出值，那么等待它是没有意义的.

Side note: if you await an observable, it will wait until the observable is complete, i.e. when it emits all its values and calls its OnComplete method. You should usually await an observable, which you know will emit only 1 value and then complete (e.g. a request to a remote server), because awaiting it will return only the last emitted value. So if your stream is expected to continuously emit values, it makes no sense to await it.

这篇关于异步方法中的 Rx StartWith 不应用起始值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！