统计实时并发流的数量

这是一个实现 CountLiveStreams 操作员:

public static IObservable<int> CountLiveStreams<T>(
    this IObservable<IObservable<T>> streamOfStreams)
{
    return streamOfStreams
        .Select(x => x.IgnoreElements().Select(_ => 0).Catch(Observable.Empty<int>())
            .Prepend(1).Append(-1))
        .Merge()
        .Scan(0, (accumulator, delta) => accumulator + delta)
        .Prepend(0);
}

每个发射的流都转换为 IObservable<int> 它发出2个值,开始时为1,结束时为-1。然后,将所有流生成的所有值对合并为一个 IObservable<int> ,最后使用 Scan 操作员。

这对我来说很有效:

public static IObservable<int> CountLiveStreams<T>(this IObservable<IObservable<T>> source) =>
    source
        .SelectMany(xs =>
            xs
                .Materialize()
                .Where(x => x.Kind != NotificationKind.OnNext)
                .Select(x => -1)
                .StartWith(1))
        .Scan((x, y) => x + y);

它使用的策略与西奥多的生产答案相同 1 和 -1 然后使用 .Scan 创建一个连续的统计数据,但我认为打电话给 .Materialize() .