This is a really inefficient operator! The use case requires this behavior, but every single `onNext` will cause a new `Observable`, `Subscriber`, and `CompositeSubscription`. Not pretty.