使用 rxcpp 进行调度和超时处理答案

【问题标题】：Scheduling and Timeout handling with rxcpp使用 rxcpp 进行调度和超时处理
【发布时间】：2015-09-23 12:00:01
【问题描述】：

我不熟悉使用 rxcpp 并尝试在以下场景中将某些功能组合在一起：

我有一个数据源可以从一个单独的源检索命令，我正在编写的代码会将这些命令检索到一个 rxcpp observable 中。它有一个特殊条件，如果在一定时间内没有收到命令，订阅者会运行 onError 函数而不是 onNext，但超时只能发生在收到第一个命令之前。在收到第一个命令后，无论需要多长时间才能收到任何进一步的命令，都不会发生超时。

我正在尝试通过以下方式完成此操作：

auto timeout = rxcpp::observable<>::timer(std::chrono::steady_clock::now() + timeout,
                             rxcpp::observe_on_event_loop()).map([](int val) // Note, converts the value type of the timer observable and converts timeouts to errors
{
    std::cout << "TIMED OUT!" << std::endl;
    throw std::runtime_error("timeout");
    return command_type();
});
auto commands = timeout.amb(rxcpp::observe_on_event_loop(), createCommandSource(event_loop_scheduler, ...));

我遇到的问题是超时发生在收到任何命令之前，即使它们在超时发生之前很远。我已经尝试过从 1000 毫秒到 5000 毫秒的超时，这没有什么区别。但是，如果我删除超时代码，则会立即收到命令。我怀疑我很可能只是误解了如何在 rxcpp 中使用调度程序，所以我想知道如何实现这一点。

【问题讨论】：

嗨！我需要看看 createCommandSource 如何创建它返回的 observable

标签： c++ reactive-programming rxcpp

【解决方案1】：

我写了一个简单的createCommandSource。这对我有用：

#include "rxcpp/rx.hpp"
using namespace rxcpp;
using namespace rxcpp::sources;
using namespace rxcpp::util;

using namespace std;

struct command_type {};

int main()
{
    auto eventloop = rxcpp::observe_on_event_loop();
    auto createCommandSource = [=]() {
        return rxcpp::observable<>::interval(std::chrono::seconds(1), eventloop).map([](long) {return command_type(); });
    };
    auto timeout = rxcpp::observable<>::timer(eventloop.now() + std::chrono::seconds(2), eventloop).map([](long ) // Note, converts the value type of the timer observable and converts timeouts to errors
    {
        std::cout << "TIMED OUT!" << std::endl;
        throw std::runtime_error("timeout");
        return command_type();
    });
    auto commands = timeout.amb(eventloop, createCommandSource().take(5));

    commands
        .as_blocking().subscribe(
        [](command_type) {printf("command\n"); },
        [](std::exception_ptr) {printf("execption\n"); });

    std::this_thread::sleep_for(std::chrono::seconds(2));

    return 0;
}

【讨论】：

【解决方案2】：

现在为了管理超时，您可以使用 .timeout() 运算符提供间隔持续时间作为参数，如下所示：

createCommandSource(event_loop_scheduler, ...).timeout((std::chrono::seconds(2));

据我了解，此运算符可确保在创建流和第一个命令、任何命令对之间以及最后一个命令和命令流的 on_complete() 事件之间超时。

【讨论】：