Расширяемая комбинация наблюдаемых - PullRequest
Новая
       44

Расширяемая комбинация наблюдаемых

0 голосов
/ 17 сентября 2018

Я бы хотел хранить списки IObservables в контейнере и подписываться на комбинацию этих наблюдаемых, получая объединенный список.Тогда я хотел бы иметь возможность добавлять больше Observables без необходимости продлевать подписку и при этом получать новые результаты.В идеале он также должен срабатывать при добавлении нового наблюдаемого в магазин.Следующий код должен объяснить больше: 

using System;
using System.Collections.Generic;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Linq;

namespace dynamic_combine
{
    class ObservableStuff
    {
        private List<IObservable<List<String>>> _listOfObservables = new List<IObservable<List<String>>>();

        public ObservableStuff() { }

        public void AddObservable(IObservable<List<String>> obs)
        {
            _listOfObservables.Add(obs);
        }

        public IObservable<IList<String>> GetCombinedObservable()
        {
            return Observable.CombineLatest(_listOfObservables)
                .Select((all) =>
                {
                    List<String> mergedList = new List<String>();
                    foreach(var list in all)
                    {
                        mergedList = mergedList.Concat(list).ToList();
                    }
                    return mergedList;
                });
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            ObservableStuff Stuff = new ObservableStuff();
            BehaviorSubject<List<String>> A = new BehaviorSubject<List<String>>(new List<String>() { "a", "b", "c" });
            BehaviorSubject<List<String>> B = new BehaviorSubject<List<String>>(new List<String>() { "d", "e", "f" });
            BehaviorSubject<List<String>> C = new BehaviorSubject<List<String>>(new List<String>() { "x", "y", "z" });

            Stuff.AddObservable(A.AsObservable());
            Stuff.AddObservable(B.AsObservable());

            Stuff.GetCombinedObservable().Subscribe((x) =>
            {
                Console.WriteLine(String.Join(",", x));
            });

            // Initial Output: a,b,c,d,e,f

            A.OnNext(new List<String>() { "1", "2", "3", "4", "5" });
            // Output: 1,2,3,4,5,d,e,f

            B.OnNext(new List<String>() { "6", "7", "8", "9", "0" });
            // Output: 1,2,3,4,5,6,7,8,9,0

            Stuff.AddObservable(C.AsObservable());
            // Wishful Output: 1,2,3,4,5,6,7,8,9,0,x,y,z

            C.OnNext(new List<String>() { "y", "e", "a", "h" });
            // Wishful Output: 1,2,3,4,5,6,7,8,9,0,y,e,a,h

            Console.WriteLine("Press the any key...");
            Console.ReadKey();
        }
    }
}

Хотя пример приведен на C #, его, наконец, необходимо реализовать в rxCpp.Также было бы интересно увидеть реализацию в других вариациях Rx.

Я установил репозиторий для проверки кода и, вероятно, расширю его на другие языки: https://gitlab.com/dwaldorf/rx-examples

BR,Daniel

1 Ответ

0 голосов
/ 18 сентября 2018

Во-первых, пара меняется, потому что ваш код не так легко прочитать.GetCombinedObservable можно переписать так: 

public IObservable<IList<String>> GetCombinedObservable()
{
    return Observable.CombineLatest(_listOfObservables)
        .Select(l => l.SelectMany(s => s).ToList());
}

Ваша проблема сводится к двум вещам: вы хотите, чтобы _listOfObservables был динамическим, что означает изменение его с List<IObservable<T>> на IObservable<IObservable<T>>,Однако проблема в том, что CombineLatest не поддерживает IObservable<IObservable<T>>, поэтому нам придется его создать.

Это приносит нам эту забавную, уродливую маленькую функцию (использует пакет nuget System.Collections.Immutable): 

public static class X
{
    public static IObservable<List<T>> DynamicCombineLatest<T>(this IObservable<IObservable<T>> source)
    {
        return source
            .SelectMany((o, i) => o.Select(item => (observableIndex: i, item: item)))
            .Scan(ImmutableDictionary<int, T>.Empty, (state, t) => state.SetItem(t.observableIndex, t.item))
            .Select(dict => dict.OrderBy(kvp => kvp.Key).Select(kvp => kvp.Value).ToList());
    }
}

Теперь мы можем обновить ваш класс: 

class ObservableStuff
{
    private ReplaySubject<IObservable<List<String>>> _subject = new ReplaySubject<IObservable<List<String>>>(int.MaxValue);

    public ObservableStuff() { }

    public void AddObservable(IObservable<List<String>> obs)
    {
        _subject.OnNext(obs);
    }

    public IObservable<IList<String>> GetCombinedObservable()
    {
        return _subject
            .DynamicCombineLatest()
            .Select(l => l.SelectMany(s => s).ToList());
    }
}
...