在 rxcpp 中创建自定义运算符

Question

我正在尝试学习如何在 rxcpp 中创建自定义运算符，并且我能够创建位于此处的运算符。但是，我想学习如何创建更多实现rxo：：operator_base并使用lift运算符的通用运算符。是否有任何文档可以通过一些简单的示例来学习这一点？

Answer 1

以下是使用 rxcpp v2 可观察提升函数的一种方法：

class MyTestOp //: public rxcpp::operators::operator_base<int>
{
public:
MyTestOp(){}
~MyTestOp(){}
rxcpp::subscriber<int> operator() (rxcpp::subscriber<int> s) const {
return rxcpp::make_subscriber<int>([s](const int & next) {
s.on_next(std::move(next + 1));
},  [&s](const std::exception_ptr & e) {
s.on_error(e);
},  [&s]() {
s.on_completed();
});
}
};

int main()
{
auto keys = rxcpp::observable<>::create<int>(
[](rxcpp::subscriber<int> dest){
for (;;) {
int key = std::cin.get();
dest.on_next(key);
}
}).
publish();
keys.lift<int>(MyTestOp()).subscribe([](int key){ 
std::cout << key << std::endl;
});
// same as use class
//keys.lift<int>([](rxcpp::subscriber<int> s) { 
//    return rxcpp::make_subscriber<int>([s](const int & next) {
//        s.on_next(std::move(next + 1));
//    },  [&s](const std::exception_ptr & e) {
//        s.on_error(e);
//    },  [&s]() {
//        s.on_completed();
//    });
//}).subscribe([](int key){ 
//    std::cout << key << std::endl;
//});
// run the loop in create
keys.connect();
return 0;
}

由于它基于模板检查，因此您不需要从任何接口继承，只需像以前一样实现运算符()即可。

而且我认为作者更希望您在评论中使用这种方式。

也许我应该使用有订阅检查...无论如何。。。

if(!s.isUnsubscribed()) { /*call s.on_xxx*/ }

Answer 2

我发现Kirk的2016年演示文稿中的以下幻灯片非常有帮助，尽管它是关于rxcppv3而不是v2的。

序列概念

struct observable {
void bind(observer);
};
struct observer {
template<class T>
void next(T);
};
struct lifter {
observer lift(observer);
};

序列实现

const auto ints = [](auto first, auto last){
return make_observable([=](auto r){ // Define observable::bind
for(auto i = first;; ++i){
r.next(i);
if (i == last) break;
}
});
};
const auto copy_if = [](auto pred){
return make_lifter([=](auto r){
return make_observer(r, [=](auto& r, auto v){ // Define observer::next
if (pred(v)) r.next(v);
});
});
};