AbstractOnSubscribe to help implement backpressure-respecting

observables.

Proposed implementation for #1930. See tests for use cases.