Added create with initial capacity.

Author: akarnokd

rxjava-core/src/main/java/rx/subjects/ReplaySubject.java

@@ -50,10 +50,12 @@
  * @param <T>
  */
 public final class ReplaySubject<T> extends Subject<T, T> {
-
     public static <T> ReplaySubject<T> create() {
+        return create(16);
+    }
+    public static <T> ReplaySubject<T> create(int initialCapacity) {
         final SubjectSubscriptionManager<T> subscriptionManager = new SubjectSubscriptionManager<T>();
-        final ReplayState<T> state = new ReplayState<T>();
+        final ReplayState<T> state = new ReplayState<T>(initialCapacity);
 
         OnSubscribeFunc<T> onSubscribe = subscriptionManager.getOnSubscribeFunc(
                 /**
@@ -91,9 +93,13 @@ public void call(SubjectObserver<? super T> o) {
 
     private static class ReplayState<T> {
         // single-producer, multi-consumer
-        final History<T> history = new History<T>();
+        final History<T> history;
         // each Observer is tracked here for what events they have received
-        final ConcurrentHashMap<Observer<? super T>, Integer> replayState = new ConcurrentHashMap<Observer<? super T>, Integer>();
+        final ConcurrentHashMap<Observer<? super T>, Integer> replayState;
+        public ReplayState(int initialCapacity) {
+            history = new History<T>(initialCapacity);
+            replayState = new ConcurrentHashMap<Observer<? super T>, Integer>();
+        }
     }
 
     private final SubjectSubscriptionManager<T> subscriptionManager;
@@ -197,10 +203,14 @@ private static <T> int replayObserverFromIndex(History<T> history, Integer l, Su
      * @param <T>
      */
     private static class History<T> {
-        private AtomicInteger index = new AtomicInteger(0);
-        private final ArrayList<T> list = new ArrayList<T>(/* 1024 */);
-        private AtomicReference<Notification<T>> terminalValue = new AtomicReference<Notification<T>>();
-
+        private final AtomicInteger index;
+        private final ArrayList<T> list;
+        private final AtomicReference<Notification<T>> terminalValue;
+        public History(int initialCapacity) {
+             index = new AtomicInteger(0);
+             list = new ArrayList<T>(initialCapacity);
+             terminalValue = new AtomicReference<Notification<T>>();
+        }
         public boolean next(T n) {
             if (terminalValue.get() == null) {
                 list.add(n);

rxjava-core/src/main/java/rx/subjects/SubjectSubscriptionManager.java

@@ -15,7 +15,6 @@
  */
 package rx.subjects;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -52,7 +51,7 @@ public Subscription onSubscribe(Observer<? super T> actualObserver) {
                     onSubscribe.call(observer);
                 }
 
-                State<T> current = null;
+                State<T> current;
                 State<T> newState = null;
                 boolean addedObserver = false;
                 Subscription s;
@@ -107,7 +106,7 @@ public void unsubscribe() {
     }
 
     protected void terminate(Action1<Collection<SubjectObserver<? super T>>> onTerminate) {
-        State<T> current = null;
+        State<T> current;
         State<T> newState = null;
         do {
             current = state.get();
@@ -133,21 +132,6 @@ protected void terminate(Action1<Collection<SubjectObserver<? super T>>> onTermi
             newState.terminationLatch.countDown();
         }
     }
-
-    /**
-     * Current snapshot of 'state.observers.keySet()' so that concurrent modifications aren't included.
-     * 
-     * This makes it behave deterministically in a single-threaded execution when nesting subscribes.
-     * 
-     * In multi-threaded execution it will cause new subscriptions to wait until the following onNext instead
-     * of possibly being included in the current onNext iteration.
-     * 
-     * @return List<Observer<T>>
-     */
-    private Collection<SubjectObserver<? super T>> snapshotOfObservers() {
-        // had to circumvent type check, we know what the array contains
-        return (Collection)state.get().observersList;
-    }
     /**
      * Returns the array of observers directly.
      * <em>Don't modify the array!</em>