c++11 - RXCPP: ブロッキング関数のタイムアウト

Question

ブロッキング関数を考えてみましょう: this_thread::sleep_for(milliseconds(3000));

私は次の動作を取得しようとしています:

Trigger Blocking Function               

|---------------------------------------------X

ブロッキング機能をトリガーしたいのですが、時間がかかりすぎる (2 秒以上) 場合は、タイムアウトする必要があります。

私は次のことをしました：

my_connection = observable<>::create<int>([](subscriber<int> s) {
    auto s2 = observable<>::just(1, observe_on_new_thread()) |
    subscribe<int>([&](auto x) {
        this_thread::sleep_for(milliseconds(3000));
        s.on_next(1);
    });
}) |
timeout(seconds(2), observe_on_new_thread());

私はこれを機能させることができません。まず、s は別のスレッドから on_next できないと思います。

だから私の質問は、これを行う正しい反応的な方法は何ですか? rxcpp でブロッキング関数をラップしてタイムアウトを追加するにはどうすればよいですか?

続いて、次のように動作する RX ストリームを取得したいと考えています。

Trigger                Cleanup

|------------------------X
                           (Delay)   Trigger           Cleanup
                                       |-----------------X