如何获取IObservable<IObservable<T>>的最新更改事件？

Question

我的系统有很多状态对象 - 连接状态、CPU负载、已登录的用户等等。所有这些事件都合并到一个可观察的流中。

我想创建一个管理实用程序，显示系统的实际状态，并显示所有计数器的值。

如何创建一个可观察的流，其中包含所有计数器最后更改的值的列表？

以下是我想要的弹珠图：

s1 (cpu):               -s1_v1----s1_v1---s1_v2
s2 (users count):       --s2_v1--s2_v1---------s2_v2
s3 (some cat purr/sec)  ----s3_v1----s3_v1----s3_v1

flatten sequence: s1_v1-s2_v1-s3_v1-s2_v1-s1_v1-s3_v1-s1_v2-s3_v1-s2_v2

期望输出：

s1_v1|s1_v1|s1_v1|s1_v2|s1_v2
      s2_v1|s2_v1|s2_v1|s2_v2
            s3_v1|s3_v1|s3_v1

目前我能做到的是这个实现：

public class StatusImplementation
{
    public static IObservable<IDictionary<TKey, TValue>> Status<TKey, TValue>(
        params IObservable<KeyValuePair<TKey, TValue>>[] observables)
    {
        var uniqueObservables = observables
            .Select(x => x.Publish().RefCount().DistinctUntilChanged());

        return Observable.Create<IDictionary<TKey, TValue>>(o =>
        {
            var compositeDisposable = new CompositeDisposable();
            var dictionary = new Dictionary<TKey, TValue>();

            foreach (var uniqueObservable in uniqueObservables)
            {
                var disposable = uniqueObservable.Subscribe(x =>
                {
                    if (dictionary.ContainsKey(x.Key) && !dictionary[x.Key].Equals(x.Value))
                    {
                        var newDictionary = new Dictionary<TKey, TValue>(dictionary);
                        newDictionary[x.Key] = x.Value;
                        dictionary = newDictionary;
                    }
                    else
                    {
                        dictionary.Add(x.Key, x.Value);
                    }

                    o.OnNext(dictionary);
                });
                compositeDisposable.Add(disposable);
            }

            return compositeDisposable;
        });
    }
}

以下是一个使用示例：

        var f1 = Observable.Interval(TimeSpan.FromMilliseconds(1000))
            .Select(x => new KeyValuePair<string, long>("event 1", x));
        var f2 = Observable.Interval(TimeSpan.FromMilliseconds(1200))
            .Select(x => new KeyValuePair<string, long>("event 2", x));
        var f3 = Observable.Interval(TimeSpan.FromMilliseconds(1250))
            .Select(x => new KeyValuePair<string, long>("event 3", x));

        var combined = f1.Merge(f2).Merge(f3);

        StatusImplementation.Status(f1, f2, f3)
            .Select(x => string.Join(", ", x.ToList()))
            .Dump("\tstatus");

        combined.Dump("normal");

以下是“转储”函数（摘自Lee Campbell的精彩图书）：

    public static void Dump<T>(this IObservable<T> source, string name)
    {
        source.Subscribe(
            i => Console.WriteLine("{0}-->{1}", name, i),
            ex => Console.WriteLine("{0} failed-->{1}", name, ex.Message),
            () => Console.WriteLine("{0} completed", name));
    }

所以问题是：有没有更好的实现这个功能的方法？可能不要在observable内部使用字典？

谢谢。

- Dmitrii

Answer 1

你可以使用 Observable.CombineLatest，它会在每次有新值到达时从所有 observables 中发出最新的值。这样你就不需要使用字典了。

 var f1 = Observable.Interval(TimeSpan.FromMilliseconds(1000))
 .Select(x = > new KeyValuePair < string, long > ("event 1", x));
var f2 = Observable.Interval(TimeSpan.FromMilliseconds(1200))
 .Select(x = > new KeyValuePair < string, long > ("event 2", x));
var f3 = Observable.Interval(TimeSpan.FromMilliseconds(1250))
 .Select(x = > new KeyValuePair < string, long > ("event 3", x));

var combined = f1.Merge(f2).Merge(f3);

Observable.CombineLatest(f1, f2, f3)
 .Select(x = > string.Join(", ", x.ToList()))
 .Dump("\tstatus");

combined.Dump("normal");

- NeddySpaghetti

---

嗨，谢谢你的想法，但不幸的是那并不符合我的要求 - CombineLatest 只会在所有 IObservable 至少发布一次后才会触发。我希望在任何 IObservables 更新后立即触发 OnNext。 - Dmitrii

你可以尝试在每个 Observable 上添加 StartWith 方法，这样它们就会有一个初始值（可能是默认值）。然后使用 CombineLatest 就能得到你想要的输出。 - Tamir Dresher

Answer 2

因此，如果您从任意数量的源observable生成combined observable开始，那么您可以执行以下操作：

var query =
    combined
        .Scan(
            new Dictionary<string, long>() as IDictionary<string, long>,
            (d, kvp) =>
            {
                var d2 = new Dictionary<string, long>(d) as IDictionary<string, long>;
                d2[kvp.Key] = kvp.Value;
                return d2;
            });

这将返回一系列字典对象，每个对象对应combined可观测对象产生的每个值。每个字典对象都是独立的实例 - 如果返回相同的实例，则会出现不断变化的值，可能会导致线程问题。

- Enigmativity