使用 rxcpp 对聚合函数进行分组？

Question

我正在尝试了解 RxCpp 的要点，它是 Microsoft 响应式扩展的本机 cpp 实现，看看我是否可以在项目中使用它，但我无法理解这些概念。

如果我有一个具有以下结构的可观察模板：

struct Person
{
    std::string name;
    std::string sex;
    int age;
}

我如何创建另一个可观察对象，其中包含按性别分组的人数、所有事件的最小年龄、最大年龄和平均年龄？

我看过示例，但看不出如何一次获得多个聚合。

Answer 1

使用 group_by 按性别划分，然后组合 min/max/average reducers 以产生每个性别所需的输出。

Updated with count, output and additional comments

这对我有用：

#include "rxcpp/rx.hpp"
using namespace rxcpp;
using namespace rxcpp::sources;
using namespace rxcpp::subjects;
using namespace rxcpp::util;

using namespace std;

struct Person
{
    string name;
    string gender;
    int age;
};

int main()
{
    subject<Person> person$;

    // group ages by gender
    auto agebygender$ = person$.
        get_observable().
        group_by(
            [](Person& p) { return p.gender;},
            [](Person& p) { return p.age;});

    // combine min max and average reductions.
    auto result$ = agebygender$.
        map([](grouped_observable<string, int> gp$){
            // the function passed to combine_latest 
            // will be called once all the source streams
            // (count, min, max, average) have produced a 
            // value. in this case, all the streams are reducers
            // that produce only one value when gp$ completes.
            // thus the function is only called once per gender 
            // with the final value of each stat.
            return gp$.
                count().
                combine_latest(
                    [=](int count, int min, int max, double average){
                        return make_tuple(gp$.get_key(), count, min, max, average);
                    },
                    gp$.min(),
                    gp$.max(),
                    gp$.map([](int age) -> double { return age;}).average());
        }).
        // this map() returns observable<observable<tuple<string, int, int, int, double>>>
        // the merge() returns observable<tuple<string, int, int, int, double>>
        // a grouped observable is 'hot' if it is not subscribed to immiediatly (in this case by merge)
        // then the values sent to it are lost.
        merge();

    // display results
    result$.
        subscribe(apply_to([](string gender, int count, int min, int max, double avg){
            cout << gender << ": count = " << count << ", range = [" << min << "-" << max << "], avg = " << avg << endl;
        }));

    //provide input data
    observable<>::from(
        Person{"Tom", "Male", 32},
        Person{"Tim", "Male", 12},
        Person{"Stel", "Other", 42},
        Person{"Flor", "Female", 24},
        Person{"Fran", "Female", 97}).
        subscribe(person$.get_subscriber());

    return 0;
}

结果输出

Female: count = 2, range = [24-97], avg = 60.5
Male: count = 2, range = [12-32], avg = 22
Other: count = 1, range = [42-42], avg = 42

使用 rxcpp 对聚合函数进行分组？

Groups aggregate functions using rxcpp?

c++

reactive-programming

rxcpp