如何在可观察量列表中使用 RxCpp 运算符?

Question

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

有没有办法在可观察量列表中使用这些运算符？ 提前谢谢。

Answer 1

这是一个幼稚的尝试，它迭代地压缩向量中的每个可观察对象并累积结果：

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

输出：

1 6 11
2 7 12
3 8 13
OnCompleted

Answer 2

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

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: %dn", v);},
[](){printf("OnCompletedn");});