Fix Concat Breaks with Double onCompleted

Fixes ReactiveX#1818
abersnaze · Nov 11, 2014 · fabcda7 · fabcda7
1 parent 71bb38b
commit fabcda7
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 4 deletions.
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorConcat.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorConcat.java
@@ -176,6 +176,10 @@ static class ConcatInnerSubscriber<T> extends Subscriber<T> {
 
         private final Subscriber<T> child;
         private final ConcatSubscriber<T> parent;
+        @SuppressWarnings("unused")
+        private volatile int once = 0;
+        @SuppressWarnings("rawtypes")
+        private final static AtomicIntegerFieldUpdater<ConcatInnerSubscriber> ONCE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(ConcatInnerSubscriber.class, "once");
 
         public ConcatInnerSubscriber(ConcatSubscriber<T> parent, Subscriber<T> child, long initialRequest) {
             this.parent = parent;
@@ -195,14 +199,18 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable e) {
-            // terminal error through parent so everything gets cleaned up, including this inner
-            parent.onError(e);
+            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+                // terminal error through parent so everything gets cleaned up, including this inner
+                parent.onError(e);
+            }
         }
 
         @Override
         public void onCompleted() {
-            // terminal completion to parent so it continues to the next
-            parent.completeInner();
+            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+                // terminal completion to parent so it continues to the next
+                parent.completeInner();
+            }
         }
 
     };

diff --git a/rxjava-core/src/test/java/rx/internal/operators/OperatorConcatTest.java b/rxjava-core/src/test/java/rx/internal/operators/OperatorConcatTest.java
@@ -695,5 +695,27 @@ public void testInnerBackpressureWithoutAlignedBoundaries() {
         ts.assertNoErrors();
         assertEquals((RxRingBuffer.SIZE * 4) + 20, ts.getOnNextEvents().size());
     }
+
+    // https://github.com/ReactiveX/RxJava/issues/1818
+    @Test
+    public void testConcatWithNonCompliantSourceDoubleOnComplete() {
+        Observable<String> o = Observable.create(new OnSubscribe<String>() {
+
+            @Override
+            public void call(Subscriber<? super String> s) {
+                s.onNext("hello");
+                s.onCompleted();
+                s.onCompleted();
+            }
+
+        });
+
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable.concat(o, o).subscribe(ts);
+        ts.awaitTerminalEvent(500, TimeUnit.MILLISECONDS);
+        ts.assertTerminalEvent();
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Arrays.asList("hello", "hello"));
+    }
 
 }