How do avoid racing conditions with is_subscribed

[daixtrose]

Hi!

I feel somewhat uncomfortable with is_subscribed. There is a sequence point after calling is_subscribed. So another thread might unsubscribe the subscriber and then s.on_next() gets called on an unsubscribed object? Is this safe? And if yes, why do we call is_subscribed anyway?

auto published_observable =
    rxcpp::sources::create<int>([](const rxcpp::subscriber<int>& s)
    {
        for (int i = 0; i < 1000; i++)
        {
            if (!s.is_subscribed())
                return;
            s.on_next(i);
        }
    })

[kirkshoop]

Yes, cross-thread cancellation is always a race. In the Rx model this is type of race is safe.

Only one call to on_error/on_completed/unsubscribe is delivered no matter how many calls are made. And once one of those calls is made no further calls to on_next will be delivered.

The subscriber uses is_subscribed() to filter calls to on_next/on_error/on_completed and calls unsubscribe when on_error/on_completed have been delivered.

calls to is_subscribed() are used to interrupt loops. unsubscribe functions passed to add() are used to teardown resources and connections. they are both needed to cover the diversity of sources. without is_subscribed() there would be so many unsubscribe functions passed to add() that just set a bool.

[daixtrose]

OK, I see I still have a lot to learn. Nonetheless my first impulse is that I would prefer an interface where this method is missing. If this type of race is safe (I assume s.on_next() will then throw or lead to a noop) why check anyway? And sorry if I do not grok it immediately.