Revert "cache now supports backpressure"

This reverts commit 18ff5af.

Author: abersnaze

src/main/java/rx/Observable.java

@@ -3594,7 +3594,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 CachedObservable.from(this);
+        return create(new OnSubscribeCache<T>(this));
     }
 
     /**
@@ -3629,7 +3629,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 capacityHint) {
-        return CachedObservable.from(this, capacityHint);
+        return create(new OnSubscribeCache<T>(this, capacityHint));
     }
 
     /**

src/main/java/rx/internal/operators/OnSubscribeCache.java

@@ -0,0 +1,76 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.internal.operators;
+
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
+import rx.subjects.ReplaySubject;
+import rx.subjects.Subject;
+
+/**
+ * This method has similar behavior to {@link Observable#replay()} except that this auto-subscribes
+ * to the source Observable rather than returning a connectable Observable.
+ * <p>
+ * <img width="640" src="https://github.com/ReactiveX/RxJava/wiki/images/rx-operators/cache.png" alt="">
+ * <p>
+ * This is useful with an Observable that you want to cache responses when you can't control the
+ * subscribe/unsubscribe behavior of all the Observers.
+ * <p>
+ * <em>Note:</em> You sacrifice the ability to unsubscribe from the origin when you use this operator, so be
+ * careful not to use this operator on Observables that emit infinite or very large numbers of
+ * items, as this will use up memory.
+ * 
+ * @param <T>
+ *            the cached value type
+ */
+public final class OnSubscribeCache<T> implements OnSubscribe<T> {
+    protected final Observable<? extends T> source;
+    protected final Subject<? super T, ? extends T> cache;
+    volatile int sourceSubscribed;
+    @SuppressWarnings("rawtypes")
+    static final AtomicIntegerFieldUpdater<OnSubscribeCache> SRC_SUBSCRIBED_UPDATER
+            = AtomicIntegerFieldUpdater.newUpdater(OnSubscribeCache.class, "sourceSubscribed");
+
+    public OnSubscribeCache(Observable<? extends T> source) {
+        this(source, ReplaySubject.<T> create());
+    }
+
+    public OnSubscribeCache(Observable<? extends T> source, int capacity) {
+        this(source, ReplaySubject.<T> create(capacity));
+    }
+
+    /* accessible to tests */OnSubscribeCache(Observable<? extends T> source, Subject<? super T, ? extends T> cache) {
+        this.source = source;
+        this.cache = cache;
+    }
+
+    @Override
+    public void call(Subscriber<? super T> s) {
+        if (SRC_SUBSCRIBED_UPDATER.compareAndSet(this, 0, 1)) {
+            source.subscribe(cache);
+            /*
+             * Note that we will never unsubscribe from 'source' unless we receive `onCompleted` or `onError`,
+             * as we want to receive and cache all of its values.
+             * 
+             * This means this should never be used on an infinite or very large sequence, similar to toList().
+             */
+        }
+        cache.unsafeSubscribe(s);
+    }
+}

src/main/java/rx/internal/util/CachedObservable.java

@@ -0,0 +1,164 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.internal.operators;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Action0;
+import rx.functions.Action1;
+import rx.functions.Func1;
+import rx.functions.Func2;
+import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
+import rx.subjects.AsyncSubject;
+import rx.subjects.BehaviorSubject;
+import rx.subjects.PublishSubject;
+import rx.subjects.ReplaySubject;
+import rx.subjects.Subject;
+
+public class OnSubscribeCacheTest {
+
+    @Test
+    public void testCache() throws InterruptedException {
+        final AtomicInteger counter = new AtomicInteger();
+        Observable<String> o = Observable.create(new Observable.OnSubscribe<String>() {
+
+            @Override
+            public void call(final Subscriber<? super String> observer) {
+                new Thread(new Runnable() {
+
+                    @Override
+                    public void run() {
+                        counter.incrementAndGet();
+                        System.out.println("published observable being executed");
+                        observer.onNext("one");
+                        observer.onCompleted();
+                    }
+                }).start();
+            }
+        }).cache();
+
+        // we then expect the following 2 subscriptions to get that same value
+        final CountDownLatch latch = new CountDownLatch(2);
+
+        // subscribe once
+        o.subscribe(new Action1<String>() {
+
+            @Override
+            public void call(String v) {
+                assertEquals("one", v);
+                System.out.println("v: " + v);
+                latch.countDown();
+            }
+        });
+
+        // subscribe again
+        o.subscribe(new Action1<String>() {
+
+            @Override
+            public void call(String v) {
+                assertEquals("one", v);
+                System.out.println("v: " + v);
+                latch.countDown();
+            }
+        });
+
+        if (!latch.await(1000, TimeUnit.MILLISECONDS)) {
+            fail("subscriptions did not receive values");
+        }
+        assertEquals(1, counter.get());
+    }
+
+    private void testWithCustomSubjectAndRepeat(Subject<Integer, Integer> subject, Integer... expected) {
+        Observable<Integer> source0 = Observable.just(1, 2, 3)
+                .subscribeOn(Schedulers.io())
+                .flatMap(new Func1<Integer, Observable<Integer>>() {
+                    @Override
+                    public Observable<Integer> call(final Integer i) {
+                        return Observable.timer(i * 20, TimeUnit.MILLISECONDS).map(new Func1<Long, Integer>() {
+                            @Override
+                            public Integer call(Long t1) {
+                                return i;
+                            }
+                        });
+                    }
+                });
+
+        Observable<Integer> source1 = Observable.create(new OnSubscribeCache<Integer>(source0, subject));
+
+        Observable<Integer> source2 = source1
+                .repeat(4)
+                .zipWith(Observable.interval(0, 10, TimeUnit.MILLISECONDS, Schedulers.newThread()), new Func2<Integer, Long, Integer>() {
+                    @Override
+                    public Integer call(Integer t1, Long t2) {
+                        return t1;
+                    }
+
+                });
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        source2.subscribe(ts);
+
+        ts.awaitTerminalEvent();
+        ts.assertNoErrors();
+        System.out.println(ts.getOnNextEvents());
+        ts.assertReceivedOnNext(Arrays.asList(expected));
+    }
+
+    @Test(timeout = 10000)
+    public void testWithAsyncSubjectAndRepeat() {
+        testWithCustomSubjectAndRepeat(AsyncSubject.<Integer> create(), 3, 3, 3, 3);
+    }
+
+    @Test(timeout = 10000)
+    public void testWithBehaviorSubjectAndRepeat() {
+        // BehaviorSubject just completes when repeated
+        testWithCustomSubjectAndRepeat(BehaviorSubject.create(0), 0, 1, 2, 3);
+    }
+
+    @Test(timeout = 10000)
+    public void testWithPublishSubjectAndRepeat() {
+        // PublishSubject just completes when repeated
+        testWithCustomSubjectAndRepeat(PublishSubject.<Integer> create(), 1, 2, 3);
+    }
+
+    @Test
+    public void testWithReplaySubjectAndRepeat() {
+        testWithCustomSubjectAndRepeat(ReplaySubject.<Integer> create(), 1, 2, 3, 1, 2, 3, 1, 2, 3, 1, 2, 3);
+    }
+
+    @Test
+    public void testUnsubscribeSource() {
+        Action0 unsubscribe = mock(Action0.class);
+        Observable<Integer> o = Observable.just(1).doOnUnsubscribe(unsubscribe).cache();
+        o.subscribe();
+        o.subscribe();
+        o.subscribe();
+        verify(unsubscribe, times(1)).call();
+    }
+}