AbstractOnSubscribe to help implement backpressure-respecting

observables.

Proposed implementation for #1930. See tests for use cases.
Such as, a `enum` for phase, and a `count` for the number of emitted items.