AbstractOnSubscribe to help build Observables one onNext at a time.

Repost of #1931.