创建一个你可以在 RxCpp 中取消订阅的 Observable

Question

我正在从严重依赖 Rx 的 C# 移植一些代码，但我很难找到一些最常用的 C# 方法的 C++ 等价物。

特别是，我想从 subscription/unsubscription 逻辑创建一个可观察对象。在 C# 中，我使用 Observable.Create<TSource> Method (Func<IObserver<TSource>, Action>) 重写来创建一个可观察对象。例如

var observable = Observable.Create<int>(observer =>
{
    observers.Add(observer);
    return () =>
    {
        observers.Remove(observer)
    };
});

是否可以对 RxCpp 做同样的事情？我认为答案在于 rx::observable<>::create(OnSubscribe os) 方法，但我不知道如何使用它来 "register" 取消订阅 lambda。

Answer 1

在 RxCpp 和 RxJava 中，.subscribe() 接受订阅者。订阅者是绑定在一起的订阅和观察者。

在 RxCpp 中，您的示例可能如下所示：

std::shared_ptr<std::list<rxcpp::subscriber<int>>> observers(new std::list<rxcpp::subscriber<int>>());

auto observable = rxcpp::observable<>::create<int>([=](rxcpp::subscriber<int> out){
    auto it = observers->insert(observers->end(), out);
    it->add([=](){
        observers->erase(it);
    });
});

注意：rxcpp::subscriber<int> 是一个隐藏观察者类型的类型遗忘器。这允许它存储在一个集合中，但为 on_next、on_error 和 on_completed.

引入了虚函数

创建一个你可以在 RxCpp 中取消订阅的 Observable

Create an Observable you can unsubscribe from in RxCpp

c#

c++

system.reactive

rxcpp