Merge pull request #1819 from benjchristensen/issuje-1818-concat

Fix Concat Breaks with Double onCompleted

Author: benjchristensen

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();
+            }
         }
 
     };

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"));
+    }
 
 }