在 rxcpp 中创建自定义运算符答案

【问题标题】：Creating custom operators in rxcpp在 rxcpp 中创建自定义运算符
【发布时间】：2018-04-10 15:08:36
【问题描述】：

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

【问题讨论】：

我看了一会儿自己，找不到任何东西。我继续执行与链接示例类似的操作，在该示例中，除了创建 operator | 的重载之外，我还包装/组合了现有的运算符来创建自己的运算符。在查看源代码后，我对现有 rxcpp 操作符的复杂性感到震惊和失望。就我个人而言，我希望下一个版本放弃对 C++11 的支持，根据 Kirk Shoop 的说法，这应该会使 rxcpp 的许多方面（包括运算符）的实现更加简单。
@RotsiserMho 我试图探讨这个话题，但观察到使用rx::operator_base 的自定义运算符需要在库代码中进行修改。在 RxCpp 的未来版本中，如果他们引入 RxJava 和 Rx.Net 中的面向接口的方法来自定义操作符，而不是仅仅坚持标准操作符的组合，这将对我们有所帮助。

标签： c++ c++11 reactive-programming rxcpp

【解决方案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;
}

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

而且我认为作者更希望你使用 cmets 中的方式。

也许我应该使用 has subscribe check ... 任何方式 ...

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

【讨论】：