c# - react 性 : Latest value to be received by IObservable

Question

我知道下面是一个阻塞调用，将返回可观察序列中的第一个值:

var result = myObservable.First();

根据我使用的主题类型，这有不同的含义:

• Subject - First() 将阻塞，直到下次调用 OnNext() 为止，这意味着最终这将是最新值
• BehaviorSubject - First() 将阻塞，直到至少一个值被推送到 OnNext() 并且因为 BehaviorSubject 跟踪最后一个值这将是最新值
• ReplaySubject - First() 将阻塞，直到至少一个值已通过 OnNext() 推送，但在许多项目已通过 OnNext 推送的情况下，它将是其缓冲区中的第一个 这不是最后一个

现在我试图找到一种一致的方法来获取最后一个值，而不管使用哪个底层 Observable。

有什么想法吗？

Answer 1

根据你的其他 Rx 问题，我认为你想要这个:

var rootSubject = new ReplaySubject<Types>();
var firstSubject = rootSubject.Where(x => x == Types.First);
var secondSubject = rootSubject.Where(x => x == Types.Second);
var thirdSubject = rootSubject.Where(x => x == Types.Third);
var forthSubject = rootSubject.Where(x => x == Types.Fourth);

var mergedSubject = Observable
              .Merge(firstSubject, secondSubject, thirdSubject, forthSubject)
        .Timeout(TimeSpan.FromSeconds(2), Observable.Return(Types.Error))    
        .Replay();

mergedSubject.Connect();

rootSubject.OnNext(Types.First);
rootSubject.OnNext(Types.Second);

var result = mergedSubject.First();

rootSubject.OnNext(Types.Third);
rootSubject.OnNext(Types.Fourth);

Console.WriteLine(String.Format("result - {0}", result));

现在不管用什么样的Subject，都返回“result - First”。

如果你想在调用 mergedSubject.First() 之前得到最新的值，你可以使用 Replay(1):

var mergedSubject = Observable
                .Merge(firstSubject, secondSubject, thirdSubject, forthSubject)
                .Timeout(TimeSpan.FromSeconds(2), Observable.Return(Types.Error))    
                .Replay(1);

在这种情况下，所有 Subject 类型都将返回“result - Second”。