Создайте наблюдаемый, который вы можете отменить в RxCpp

Question

Я переношу некоторый код из C#, который сильно зависит от Rx, и мне трудно найти эквиваленты C++ для некоторых из наиболее используемых методов C#.

В частности, я хочу создать наблюдаемое из логики подписки/отмены подписки. В 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), но я не могу понять, как его использовать, чтобы «зарегистрировать» лямбда без подписки.

Answer 1

В RxCpp и RxJava. Подписка() берет подписчика. Абонент - подписка, а наблюдатель связан вместе.

В 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> тип-forgetter, который скрывает тип наблюдателя. Это позволяет хранить его в коллекции, но вводит виртуальные функции для on_next, on_error и on_completed.