有效地将许多 IObservable<bool> 流与布尔运算符结合起来

Question

我希望组合许多 IObservable<bool> 流，以便当所有流的最新值为 true 时，发出 true，否则发出 false。

CombinedLast 可以让我轻松地为两个流构建类似的东西，但是 a) 我不确定 API 是否可以轻松组合数千个流 b) 我不确定它的效率如何即使可以。

All 有点类似于我想要的，除了我假设它适用于单个序列并且一旦 false 不能动态变回 true。

虽然DistintUntilChanged 运算符对此可能没有效率，但我还需要将值“区分到更改”？

我希望有一个 O(1) 算法。

Answer 1

结合最新的一个好方法是从IObservable<IObservable<T>> 开始，然后将其转换为IObservable<T[]>。这成为一种非常动态的方式来组合您需要的任意数量的值。

这是一个扩展方法：

public static IObservable<T[]> CombineLatest<T>(this IObservable<IObservable<T>> sources)
{
    return
        sources.Publish(ss =>
            Observable.Create<T[]>(o =>
            {
                var composite = new CompositeDisposable();
                var list = new List<T>();
                composite.Add(
                    ss.Subscribe(source =>
                    {
                        var index = list.Count;
                        list.Add(default(T));
                        composite.Add(source.Subscribe(x => list[index] = x));
                    }));
                composite.Add(ss.Merge().Select(x => list.ToArray()).Subscribe(o));
                return composite;
            }));
}

这很好地创建和跟踪所有订阅，并使用一个闭包来定义每个订阅需要使用的index 来更新其在用于输出的list 中的值。

如果你这样使用它：

var sources = new Subject<IObservable<bool>>();

var output = sources.CombineLatest();

output.Subscribe(x => Console.WriteLine(x));

var s1 = new Subject<bool>();
sources.OnNext(s1);
s1.OnNext(true);
var s2 = new Subject<bool>();
sources.OnNext(s2);
s2.OnNext(false);
var s3 = new Subject<bool>();
sources.OnNext(s3);
s3.OnNext(true);
s2.OnNext(true);
s1.OnNext(false);

然后你得到这个输出：

如果您将output 的定义更改为var output = sources.CombineLatest().Select(xs => xs.Aggregate((x, y) => x & y));，那么您会得到我认为您需要的输出：

真的 错误的 错误的 真的 错误的

Answer 2

我不知道如何以经典的功能方式做到这一点，但仍然达到 O(1)。这使用了可变状态，观察每条消息是 O(1)，但内存是 O(n)：

public IObservable<bool> CombineBooleans(this IObservable<bool>[] source)
{
    return source.Select((o, i) => o.Select(b => (value: b, index: i)))
        .Merge()
        .Scan((array: new bool[source.Length], countFalse: source.Length), (state, item) =>
        {
            var countFalse = state.countFalse;

            if (state.array[item.index] == item.value)
                return (state.array, countFalse);           //nothing to change, emit same state
            else if (state.array[item.index])               //previous/current state is true, becoming false
            {
                countFalse++;
                state.array[item.index] = false;
            }
            else                                            //previous/current state is false, becoming true
            {
                countFalse--;
                state.array[item.index] = true;
            }
            return (state.array, countFalse);
        })
        .Scan((countFalse: source.Length, oldCountFalse: source.Length), (state, item) => (countFalse: item.countFalse, oldCountFalse: state.countFalse))
        .SelectMany(state =>
            state.countFalse == 1 && state.oldCountFalse == 0
                ? Observable.Return(false)
                : state.countFalse == 0 && state.oldCountFalse == 1
                    ? Observable.Return(true)
                    : Observable.Empty<bool>()
        )
        .Publish()
        .RefCount();
}

编辑：添加.Publish().Refcount() 以消除多订阅者错误。