add protection to OrderedMerge.SourceSubscriber against multiple term…

…inal events #25
davidmoten · May 19, 2017 · 08e66ff · 08e66ff
1 parent 85ab0bb
commit 08e66ff
Showing 1 changed file with 12 additions and 1 deletion.
diff --git a/src/main/java/com/github/davidmoten/rx/internal/operators/OrderedMerge.java b/src/main/java/com/github/davidmoten/rx/internal/operators/OrderedMerge.java
@@ -17,6 +17,7 @@
 import rx.internal.operators.NotificationLite;
 import rx.internal.util.RxRingBuffer;
 import rx.internal.util.unsafe.MpscLinkedQueue;
+import rx.plugins.RxJavaHooks;
 
 /**
  * @author David Karnokd
@@ -295,7 +296,7 @@ static final class SourceSubscriber<T> extends Subscriber<T> {
         final MergeProducer<T> parent;
         volatile boolean done;
 
-        public SourceSubscriber(MergeProducer<T> parent) {
+        SourceSubscriber(MergeProducer<T> parent) {
             queue = RxRingBuffer.getSpscInstance();
             this.parent = parent;
         }
@@ -312,6 +313,9 @@ public void requestMore(long n) {
 
         @Override
         public void onNext(T t) {
+            if (done) {
+                return;
+            }
             try {
                 queue.onNext(NotificationLite.next(t));
             } catch (MissingBackpressureException mbe) {
@@ -336,12 +340,19 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable e) {
+            if (done) {
+                RxJavaHooks.onError(e);
+                return;
+            }
             done = true;
             parent.error(e);
         }
 
         @Override
         public void onCompleted() {
+            if (done) {
+                return;
+            }
             done = true;
             parent.emit();
         }