AbstractOnSubscribe to help implement backpressure-respecting

observables.

Proposed implementation for #1930. See tests for use cases.
IMO, I don't like that people have to use magic numbers in their code. How about creating an `enum` for them?