c# - react 性 : Converting merged IObservable's into one stream that acts like BehaviorSubject

Question

这是我的示例代码...

var rootSubject = new Subject<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.Forth);

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

rootSubject.OnNext(Types.Second);

var result = mergedSubject.First();

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

出于某种原因，它总是超时并返回错误类型。知道这里发生了什么吗？

我想做的是创建一个合并的 Iobservable，它是一个行为类似于 BehaviorSubject 的流，这样如果在 .First() 之前调用 .OnNext(...)，first 将具有一个值。

Answer 1

我认为问题在于您基本上没有连接回放序列 - 或者可能连接得太晚了。 (我不知道 RefCount 的细节，但我怀疑它只有在订阅它时才会连接。)

这是一个可行的替代方案:

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

mergedSubject.Connect();
rootSubject.OnNext(Types.Second);

var result = mergedSubject.First();

我不知道这是否满足您的所有需求，但它至少为您的测试代码打印了正确的结果:)