Add cache(int capacity) to Observable

Author: gliptak

rxjava-core/src/main/java/rx/Observable.java

@@ -3425,6 +3425,19 @@ public final Observable<T> cache() {
         return create(new OnSubscribeCache<T>(this));
     }
 
+    /**
+     * {@code cache} with initial capacity.
+     *
+     * @param capacity
+     *            initial cache size
+     * @return an Observable that, when first subscribed to, caches all of its items and notifications for the
+     *         benefit of subsequent subscribers
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#cache">RxJava Wiki: cache()</a>
+     */
+    public final Observable<T> cache(int capacity) {
+        return create(new OnSubscribeCache<T>(this, capacity));
+    }
+
     /**
      * Returns an Observable that emits the items emitted by the source Observable, converted to the specified
      * type.

rxjava-core/src/main/java/rx/internal/operators/OnSubscribeCache.java

@@ -52,6 +52,10 @@ 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;

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -694,6 +694,54 @@ public void call(String v) {
         assertEquals(1, counter.get());
     }
 
+    @Test
+    public void testCacheWithCapacity() throws InterruptedException {
+        final AtomicInteger counter = new AtomicInteger();
+        Observable<String> o = Observable.create(new OnSubscribe<String>() {
+
+            @Override
+            public void call(final Subscriber<? super String> observer) {
+                new Thread(new Runnable() {
+
+                    @Override
+                    public void run() {
+                        counter.incrementAndGet();
+                        observer.onNext("one");
+                        observer.onCompleted();
+                    }
+                }).start();
+            }
+        }).cache(1);
+
+        // 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);
+                latch.countDown();
+            }
+        });
+
+        // subscribe again
+        o.subscribe(new Action1<String>() {
+
+            @Override
+            public void call(String v) {
+                assertEquals("one", v);
+                latch.countDown();
+            }
+        });
+
+        if (!latch.await(1000, TimeUnit.MILLISECONDS)) {
+            fail("subscriptions did not receive values");
+        }
+        assertEquals(1, counter.get());
+    }
+
     /**
      * https://github.com/Netflix/RxJava/issues/198
      *