c++ - RxCpp : how to control subject observer's lifetime when used with 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()，但仍然无法让程序正常退出。

请注意，我在这里使用“主题”是因为“发布”可以从不同的地方调用(可以来自不同的线程)。我不确定这是否是最好的机制，我愿意接受其他方式来实现这一点。

建议？

Answer 1

这非常接近工作..

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

这是一种确保 foo 在 Data 析构后阻塞的方法。这是使用现有的数据契约(Contract)。

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");
}

或者，改变 Data 的形状(添加一个完整的方法)将允许以下代码:

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");
}

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