Fix Merge Bug

Working on #1420

Author: benjchristensen

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

@@ -53,7 +53,7 @@ private static final class MergeSubscriber<T> extends Subscriber<Observable<? ex
         private int wip;
         private boolean completed;
 
-        private SubscriptionIndexedRingBuffer<InnerSubscriber<T>> childrenSubscribers;
+        private volatile SubscriptionIndexedRingBuffer<InnerSubscriber<T>> childrenSubscribers;
 
         private RxRingBuffer scalarValueQueue = null;
 
@@ -244,11 +244,12 @@ private boolean drainQueuesIfNeeded() {
                         emitted = drainScalarValueQueue();
                         drainChildrenQueues();
                     } finally {
-                        if (!releaseEmitLock()) {
-                            return true;
-                        }
+                        boolean moreToDrain = releaseEmitLock();
                         // request outside of lock
                         request(emitted);
+                        if (!moreToDrain) {
+                            return true;
+                        }
                         // otherwise we'll loop and get whatever was added 
                     }
                 } else {
@@ -350,27 +351,29 @@ public void onCompleted() {
             }
             if (c) {
                 // complete outside of lock
-                actual.onCompleted();
+                drainAndComplete();
             }
         }
 
         void completeInner(InnerSubscriber<T> s) {
-            try {
-                boolean sendOnComplete = false;
-                synchronized (this) {
-                    wip--;
-                    if (wip == 0 && completed) {
-                        sendOnComplete = true;
-                    }
-                }
-                if (sendOnComplete) {
-                    actual.onCompleted();
+            boolean sendOnComplete = false;
+            synchronized (this) {
+                wip--;
+                if (wip == 0 && completed) {
+                    sendOnComplete = true;
                 }
-            } finally {
-                childrenSubscribers.remove(s.sindex);
+            }
+            childrenSubscribers.remove(s.sindex);
+            if (sendOnComplete) {
+                drainAndComplete();
             }
         }
 
+        private void drainAndComplete() {
+            drainQueuesIfNeeded(); // TODO need to confirm whether this is needed or not
+            actual.onCompleted();
+        }
+
     }
 
     private static final class MergeProducer<T> implements Producer {
@@ -402,9 +405,10 @@ private static final class InnerSubscriber<T> extends Subscriber<T> {
         final MergeSubscriber<T> parentSubscriber;
         final MergeProducer<T> producer;
         /** Make sure the inner termination events are delivered only once. */
-        volatile int once;
+        volatile int terminated;
         @SuppressWarnings("rawtypes")
-        static final AtomicIntegerFieldUpdater<InnerSubscriber> ONCE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(InnerSubscriber.class, "once");
+        static final AtomicIntegerFieldUpdater<InnerSubscriber> ONCE_TERMINATED = AtomicIntegerFieldUpdater.newUpdater(InnerSubscriber.class, "terminated");
+
         private final RxRingBuffer q = RxRingBuffer.getSpmcInstance();
         /* protected by emitLock */
         int emitted = 0;
@@ -425,14 +429,14 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable e) {
             // it doesn't go through queues, it immediately onErrors and tears everything down
-            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+            if (ONCE_TERMINATED.compareAndSet(this, 0, 1)) {
                 parentSubscriber.onError(e);
             }
         }
 
         @Override
         public void onCompleted() {
-            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+            if (ONCE_TERMINATED.compareAndSet(this, 0, 1)) {
                 emit(null, true);
             }
         }

rxjava-core/src/main/java/rx/internal/util/IndexedRingBuffer.java

@@ -122,21 +122,16 @@ public int add(E e) {
     }
 
     public E remove(int index) {
-        try {
-            E e;
-            if (index < SIZE) {
-                // fast-path when we are in the first section
-                e = elements.array.getAndSet(index, null);
-            } else {
-                int sectionIndex = index % SIZE;
-                e = getElementSection(index).array.getAndSet(sectionIndex, null);
-            }
-            pushRemovedIndex(index);
-            return e;
-        } catch (NullPointerException ne) {
-            ne.printStackTrace();
-            throw ne;
+        E e;
+        if (index < SIZE) {
+            // fast-path when we are in the first section
+            e = elements.array.getAndSet(index, null);
+        } else {
+            int sectionIndex = index % SIZE;
+            e = getElementSection(index).array.getAndSet(sectionIndex, null);
         }
+        pushRemovedIndex(index);
+        return e;
     }
 
     private IndexSection getIndexSection(int index) {

rxjava-core/src/main/java/rx/internal/util/RxRingBuffer.java

@@ -260,6 +260,13 @@ public int count() {
         }
         return queue.size();
     }
+    
+    public boolean isEmpty() {
+        if (queue == null) {
+            return true;
+        }
+        return queue.isEmpty();
+    }
 
     public Object poll() {
         if (queue == null) {

rxjava-core/src/main/java/rx/internal/util/SubscriptionList.java

@@ -86,6 +86,7 @@ public void unsubscribe() {
         }
         // we will only get here once
         unsubscribeFromAll(subscriptions);
+        subscriptions = null;
     }
 
     private static void unsubscribeFromAll(Collection<Subscription> subscriptions) {