RXCPP: Timeout on blocking function

Consider a blocking function: this_thread::sleep_for(milliseconds(3000));

I'm trying to get the following behavior:

Trigger Blocking Function               

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

I want to trigger the blocking function and if it takes too long (more than two seconds), it should timeout.

I've done the following:

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());

I can't get this to work. For starters, I think s can't on_next from a different thread.

So my question is, what is the correct reactive way of doing this? How can I wrap a blocking function in rxcpp and add a timeout to it?

Subsequently, I want to get an RX stream that behaves like this:

Trigger                Cleanup

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