对Subject<IEnumerable<AggregatedObj>>的反应式扩展Subject<IEnumerable<Obj>>

Question

我很难接受一个主题

public Subject<IEnumerable<Person>> PersonDataSubject;

并将其转换为：

public Subject<IEnumerable<BornInYear>> BornInYearSubject;

..。使用一些linq聚合。

下面的例子将它放在了更多的上下文中，而我正在努力解决的是如何从对PersonDataSubject的订阅中获得一个IEnumerable到BornInYearSubject中。

不管我怎么尝试，我最终得到的都是IObservable<BornInYear>，而不是IObservable<IEnumerable<BornInYear>>。

目标是让类的客户端能够订阅这两个主题，并在每次“下一步”通知时获得相应类型的IEnumerable。

public class ReactiveTest
{
    public class Person
    {
        public string name;
        public DateTime dob;
    };

    public class BornInYear
    {
        public int Year;
        public int Count;
    }

    public Subject<IEnumerable<Person>> PersonDataSubject = new Subject<IEnumerable<Person>>();
    public Subject<IEnumerable<BornInYear>> BornInYearSubject= new Subject<IEnumerable<BornInYear>>();

    public void LoadData()
    {
        // Go to hypotheritical web service and get batch of people.
        IEnumerable<Person> people = WebService.Fetch();

        // Notify subscribers we have a fresh batch of data.
        PersonDataSubject.OnNext(people);
    }

    public ReactiveTest()
    {
        // Hookup BornInYearSubject to listen to PersonDataSubject and publish the summarised data.
        PersonDataSubject.Subscribe(pd => pd.GroupBy(p => p.dob.Year)
                                            .Select(ps => new BornInYear { Year = ps.Key, Count = ps.Count()})
                                            .AsParallel()
            );

        // How do I get the results of this out and published onto BornInYearSubject?
    }
}

现在我知道我可以使用Task.Factory.StartNew(...)...作为我的PersonDataSubject订阅OnNext来实现这一点，但我相信这肯定是可能的，保持更多的反应？

Answer 1

好的，这行得通。谢谢你的点子，伙计们--事后看来，答案似乎非常明显！

using System;
using System.Collections.Generic;
using System.Linq;

namespace TestReactive
{
    public class ReactiveTest
    {
        public class Person
        {
            public string name;
            public DateTime dob;
        };

        public class BornInYear
        {
            public int Year;
            public int Count;
        }

        public Subject<IEnumerable<Person>> PersonDataSubject = new Subject<IEnumerable<Person>>();
        public Subject<IEnumerable<BornInYear>> BornInYearSubject = new Subject<IEnumerable<BornInYear>>();

        public void LoadData()
        {
            IEnumerable<Person> people = new List<Person> {
                new Person() {name = "Bill", dob = DateTime.Now.AddYears(-10)},
                new Person() {name = "Pete", dob = DateTime.Now.AddYears(-5)},
                new Person() {name = "Judy", dob = DateTime.Now.AddYears(-1)},
                new Person() {name = "Mike", dob = DateTime.Now.AddYears(-5)},
                new Person() {name = "Jake", dob = DateTime.Now.AddYears(-5)},
                new Person() {name = "Fred", dob = DateTime.Now.AddYears(-13)},
            };

            // Notify subscribers we have a fresh batch of data.
            PersonDataSubject.OnNext(people);
        }

        public ReactiveTest()
        {
            var subj = PersonDataSubject.Select(pds => pds.GroupBy(pd => pd.dob.Year)
                                                          .Select(p => new BornInYear {
                                                              Year = p.Key, Count = p.Count()
                                                          }).AsParallel());
            subj.Subscribe(BornInYearSubject);  

            BornInYearSubject.Subscribe( x=> Console.WriteLine("{0}", x.Count()));
            LoadData();
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            ReactiveTest rt = new ReactiveTest();
        }
    }
}

Answer 2

这样如何：

PersonDataSubject
    .GroupBy(x => x.Dob.Year)
    .Select(x => x.Aggregate(new List<BornInYear>(), (acc, x) => { acc.Add(new BornInYear { Year = ps.Key }); return acc; }))