Rewrite Observable.cache() & Flowable.cache() #6270
Description
The current implementations of Observable.cache() & Flowable.cache() look old and have several problems:
- They dispose the upstream when they terminate, which shouldn't happen.
- They have an extra indirection for their state but not sure why. Probably a structuring remnant from 1.x.
- The general drain logic seem to use too many variables.
PR welcome.