RxCpp：与 buffer_with_time 一起使用时如何控制对象观察者的生命周期

Question

以下代码的目的是让各种类将数据发布到可观察对象。有的类会观察每一个数据，有的会周期性的用buffer_with_time()观察。

这在程序退出之前运行良好，然后崩溃，可能是因为使用 buffer_with_time() 的观察者仍然挂在某个线程上。

struct Data
{
    Data() : _subscriber(_subject.get_subscriber()) { }
    ~Data()  { _subscriber.on_completed(); }

    void publish(std::string data) { _subscriber.on_next(data); }
    rxcpp::observable<std::string> observable() { return _subject.get_observable(); }

private:
    rxcpp::subjects::subject<std::string> _subject;
    rxcpp::subscriber<std::string> _subscriber;
};

void foo()
{
    Data data;

    auto period = std::chrono::milliseconds(30);
    auto s1 = data.observable()
        .buffer_with_time(period , rxcpp::observe_on_new_thread())
        .subscribe([](std::vector<std::string>& data) 
           { std::cout << data.size() << std::endl; });

    data.publish("test 1");
    data.publish("test 2");
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    // hope to call something here so s1's thread can be joined.
    // program crashes upon exit
}

我试过调用"s1.unsubscribe()"和各种as_blocking()、from()、merge()，但仍然无法让程序正常退出。

请注意，我在这里使用了 "subjects"，因为 "publish" 可以从不同的地方调用（可以从不同的线程）。我不确定这是否是最好的机制，我愿意接受其他方式来实现这一点。

建议？

Answer 1

这非常接近工作..

但是，让数据析构函数完成输入，同时还希望订阅在输入完成之前阻止 foo 的退出，这会使事情变得更加复杂。

这里有一种方法可以确保 foo 在 Data 析构后阻塞。这是使用现有的数据协定。

void foo1()
{
    rxcpp::observable<std::vector<std::string>> buffered;
    {
        Data data;

        auto period = std::chrono::milliseconds(30);
        buffered = data.observable()
            .buffer_with_time(period , rxcpp::observe_on_new_thread())
            .publish().ref_count();

        buffered
            .subscribe([](const std::vector<std::string>& data)
               { printf("%lu\n", data.size()); },
               [](){printf("data complete\n");});

        data.publish("test 1");
        data.publish("test 2");

        // hope to call something here so s1's thread can be joined.
        // program crashes upon exit
    }
    buffered.as_blocking().subscribe();

    printf("exit foo1\n");
}

或者，改变数据的形状（添加一个完整的方法）将允许以下代码：

struct Data
{
    Data() : _subscriber(_subject.get_subscriber()) { }
    ~Data()  { complete(); }

    void publish(std::string data) { _subscriber.on_next(data); }
    void complete() {_subscriber.on_completed();}
    rxcpp::observable<std::string> observable() { return _subject.get_observable(); }

private:
    rxcpp::subjects::subject<std::string> _subject;
    rxcpp::subscriber<std::string> _subscriber;
};

void foo2()
{
    printf("foo2\n");

    Data data;

    auto newthread = rxcpp::observe_on_new_thread();

    auto period = std::chrono::milliseconds(30);
    auto buffered = data.observable()
        .buffer_with_time(period , newthread)
        .tap([](const std::vector<std::string>& data)
           { printf("%lu\n", data.size()); },
           [](){printf("data complete\n");});

    auto emitter = rxcpp::sources::timer(std::chrono::milliseconds(0), newthread)
        .tap([&](long) {
            data.publish("test 1");
            data.publish("test 2");
            data.complete();
        });

    // hope to call something here so s1's thread can be joined.
    // program crashes upon exit
    buffered.combine_latest(newthread, emitter).as_blocking().subscribe();

    printf("exit foo2\n");
}

我认为这更好地表达了依赖关系..

RxCpp：与 buffer_with_time 一起使用时如何控制对象观察者的生命周期

RxCpp: how to control subject observer's lifetime when used with buffer_with_time

c++

reactive-programming