如何在可观察对象列表上使用 RxCpp 运算符？答案

【问题标题】：How to use RxCpp operators on a list of observables?如何在可观察对象列表上使用 RxCpp 运算符？
【发布时间】：2020-01-01 06:11:26
【问题描述】：

我想在 RxCpp observables 的列表/数组上使用 zip 运算符，但 RxCpp 运算符只接受可变参数，而且我的列表的大小是动态的（在运行时已知）。

有没有办法在 observables 列表中使用这些运算符？提前致谢。

【问题讨论】：

也值得一试：flat_map operator

标签： c++ reactive variadic rxcpp

【解决方案1】：

这是一个简单的尝试，它迭代地将每个 observable 压缩到一个向量中并累积结果：

template <typename T>
rxcpp::observable<std::shared_ptr<std::vector<T>>> zip_v(const std::vector<rxcpp::observable<T>>& observables) {
    // map the first observable to convert values to a single-element vector
    auto it = observables.cbegin();
    rxcpp::observable<std::shared_ptr<std::vector<T>>> acc = it->map([](T t) {
        return std::make_shared<std::vector<T>>(std::initializer_list<T>{ t });
    });

    // fold each observable into accumulator by zipping, and pushing back value
    while (++it != observables.cend()) {
        acc = acc.zip([](std::shared_ptr<std::vector<T>> acc, T next) { 
            acc->push_back(next);
            return acc;
        }, *it);
    }
    return acc;
}

使用示例：

    std::vector<rxcpp::observable<int>> sources{
        rxcpp::observable<>::range(1, 5),
        rxcpp::observable<>::range(6, 10),
        rxcpp::observable<>::range(11, 15)
    };
    zip_v(sources).
        take(3).
        subscribe(
            [](auto vs){
                for (auto& v : *vs) {
                    printf("%d ", v);
                }
                printf("\n");
            },
            [](){ printf("complete\n"); });

输出：

    1 6 11
    2 7 12
    3 8 13
    OnCompleted

【讨论】：

【解决方案2】：

在RxCpp的github页面上发布这个问题后，有人建议我使用iterate函数。

std::array< rxcpp::observable<int>, 3 > a={{rxcpp::just(1), rxcpp::just(2), rxcpp::just(3)}};
auto values = rxcpp::observable<>::iterate(a);
values.
    merge().
    subscribe(
        [](int v){printf("OnNext: %d\n", v);},
        [](){printf("OnCompleted\n");});

【讨论】：