Merge pull request #2969 from akarnokd/OperatorCacheRewrite2

Operator cache() now supports backpressure

Author: akarnokd

src/main/java/rx/Observable.java

@@ -3504,7 +3504,7 @@ public final <B> Observable<List<T>> buffer(Observable<B> boundary, int initialC
      * @see <a href="http://reactivex.io/documentation/operators/replay.html">ReactiveX operators documentation: Replay</a>
      */
     public final Observable<T> cache() {
-        return create(new OnSubscribeCache<T>(this));
+        return CachedObservable.from(this);
     }
 
     /**
@@ -3539,7 +3539,7 @@ public final Observable<T> cache() {
      * @see <a href="http://reactivex.io/documentation/operators/replay.html">ReactiveX operators documentation: Replay</a>
      */
     public final Observable<T> cache(int capacity) {
-        return create(new OnSubscribeCache<T>(this, capacity));
+        return CachedObservable.from(this, capacity);
     }
 
     /**