[yukiendo-pfr]

subscribe() and create_timer() hand the subscription / timer to the returned stream (stream->sub_, stream->timer_), while the callback they register captures the same stream as a shared_ptr. The two keep each other alive, so dropping the stream never releases them: the subscription stays registered and keeps queueing messages nobody reads, and the timer keeps firing. poll_until() creates a timer on every wait that does not succeed at once, so wait_for_service() / wait_for_action() leave one behind each time they have to wait.

The callbacks now hold the stream weakly. Dropping the stream (e.g. at the end of the coroutine that created it, as in the guide's examples) releases the subscription or timer with it; a coroutine awaiting next() still owns the stream, so it is not released under it.

Tests added:

• TopicStreamTest.DroppingTheStreamUnsubscribes: the stream expires and count_subscribers() drops back to 0
• TimerStreamTest.DroppingTheStreamReleasesTheTimer
• WaitForServiceTest.TimeoutLeavesNoTimerBehind: no timer is left on the node after a timed-out wait_for_service()

All three fail without the change and pass with it.

[otamachan]

LGTM