c++ - 在 rxcpp 中创建自定义运算符

Question

我正在尝试学习如何在 rxcpp 中创建自定义运算符，并且我能够创建位于 here 中的运算符.但是，我想了解如何创建更多通用运算符来实现 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;
}

因为它是基于模板检查的，所以你不需要继承任何接口(interface)，像以前一样实现一个operator()就可以了。

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

也许我应该使用 has subscribe 检查...无论如何...

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