反应式扩展 主题<IEnumerable<Obj>> to Subject<IEnumerable<AggregatedObj>>

Question

我在选修科目时遇到问题

public Subject<IEnumerable<Person>> PersonDataSubject;

并将其转换为：

public Subject<IEnumerable<BornInYear>> BornInYearSubject;

使用一些linq聚合。

下面的例子把它放在了更多的上下文中，我正在努力解决如何从对PersonDataSubject的订阅中将IEnumerable获取到BornYearSubject中。

无论我尝试什么，我最终都会得到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

怎么样：

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

Answer 2

好的，这样可以。谢谢你们的想法——事后看来，答案似乎非常明显！

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();
        }
    }
}