using System.Reactive.Linq; using System.Reactive.Subjects; namespace InnovEnergy.Lib.Utils; public static class DictObservables { public static IObservable> DictObserve ( this IObservable> src, Func, IObservable> map ) where K : notnull { var subjects = new Dictionary>(); var results = new Dictionary(); return src.Select(Map); // TODO: thread safety IReadOnlyDictionary Map(IReadOnlyDictionary inDict) { results.Clear(); foreach (var (key, value) in inDict) { if (!subjects.ContainsKey(key)) { var subject = new Subject(); map(subject).Subscribe(r => results[key] = r); subjects[key] = subject; } subjects[key].OnNext(value); } return results; } } public static IObservable> DictWhere ( this IObservable> src, Func predicate ) where K : notnull { return from dict in src select dict .Where(e => predicate(e.Value)) .ToDictionary(e => e.Key, e => e.Value); } public static IObservable> DictSelect ( this IObservable> src, Func map ) where K : notnull { return from dict in src select dict.ToDictionary(e => e.Key, e => map(e.Value)); } public static IObservable> DictSelect ( this IObservable> src, Func, KeyValuePair> map ) where K1 : notnull where K2 : notnull { return from dict in src select dict .Select(map) .ToDictionary(e => e.Key, e => e.Value); } }