Merge pull request #4849 from akarnokd/CompletableConcatFix

1.x: fix Completable.concat & merge hanging in async situations

Author: akarnokd

src/main/java/rx/internal/operators/CompletableOnSubscribeConcat.java

@@ -21,9 +21,9 @@
 import rx.*;
 import rx.Completable.OnSubscribe;
 import rx.exceptions.MissingBackpressureException;
+import rx.internal.subscriptions.SequentialSubscription;
 import rx.internal.util.unsafe.SpscArrayQueue;
 import rx.plugins.RxJavaHooks;
-import rx.subscriptions.SerialSubscription;
 
 public final class CompletableOnSubscribeConcat implements OnSubscribe {
     final Observable<Completable> sources;
@@ -39,30 +39,29 @@ public CompletableOnSubscribeConcat(Observable<? extends Completable> sources, i
     public void call(CompletableSubscriber s) {
         CompletableConcatSubscriber parent = new CompletableConcatSubscriber(s, prefetch);
         s.onSubscribe(parent);
-        sources.subscribe(parent);
+        sources.unsafeSubscribe(parent);
     }
 
     static final class CompletableConcatSubscriber
     extends Subscriber<Completable> {
         final CompletableSubscriber actual;
-        final SerialSubscription sr;
+        final SequentialSubscription sr;
 
         final SpscArrayQueue<Completable> queue;
 
-        volatile boolean done;
+        final ConcatInnerSubscriber inner;
 
         final AtomicBoolean once;
 
-        final ConcatInnerSubscriber inner;
+        volatile boolean done;
 
-        final AtomicInteger wip;
+        volatile boolean active;
 
         public CompletableConcatSubscriber(CompletableSubscriber actual, int prefetch) {
             this.actual = actual;
             this.queue = new SpscArrayQueue<Completable>(prefetch);
-            this.sr = new SerialSubscription();
+            this.sr = new SequentialSubscription();
             this.inner = new ConcatInnerSubscriber();
-            this.wip = new AtomicInteger();
             this.once = new AtomicBoolean();
             add(sr);
             request(prefetch);
@@ -74,9 +73,7 @@ public void onNext(Completable t) {
                 onError(new MissingBackpressureException());
                 return;
             }
-            if (wip.getAndIncrement() == 0) {
-                next();
-            }
+            drain();
         }
 
         @Override
@@ -94,9 +91,7 @@ public void onCompleted() {
                 return;
             }
             done = true;
-            if (wip.getAndIncrement() == 0) {
-                next();
-            }
+            drain();
         }
 
         void innerError(Throwable e) {
@@ -105,32 +100,45 @@ void innerError(Throwable e) {
         }
 
         void innerComplete() {
-            if (wip.decrementAndGet() != 0) {
-                next();
-            }
-            if (!done) {
-                request(1);
-            }
+            active = false;
+            drain();
         }
 
-        void next() {
-            boolean d = done;
-            Completable c = queue.poll();
-            if (c == null) {
-                if (d) {
-                    if (once.compareAndSet(false, true)) {
-                        actual.onCompleted();
-                    }
-                    return;
-                }
-                RxJavaHooks.onError(new IllegalStateException("Queue is empty?!"));
+        void drain() {
+            ConcatInnerSubscriber inner = this.inner;
+            if (inner.getAndIncrement() != 0) {
                 return;
             }
 
-            c.unsafeSubscribe(inner);
+            do {
+                if (isUnsubscribed()) {
+                    return;
+                }
+                if (!active) {
+                    boolean d = done;
+                    Completable c = queue.poll();
+                    boolean empty = c == null;
+
+                    if (d && empty) {
+                        actual.onCompleted();
+                        return;
+                    }
+
+                    if (!empty) {
+                        active = true;
+                        c.subscribe(inner);
+
+                        request(1);
+                    }
+                }
+            } while (inner.decrementAndGet() != 0);
         }
 
-        final class ConcatInnerSubscriber implements CompletableSubscriber {
+        final class ConcatInnerSubscriber
+        extends AtomicInteger
+        implements CompletableSubscriber {
+            private static final long serialVersionUID = 7233503139645205620L;
+
             @Override
             public void onSubscribe(Subscription d) {
                 sr.set(d);

src/main/java/rx/internal/operators/CompletableOnSubscribeMerge.java

@@ -43,7 +43,7 @@ public CompletableOnSubscribeMerge(Observable<? extends Completable> source, int
     public void call(CompletableSubscriber s) {
         CompletableMergeSubscriber parent = new CompletableMergeSubscriber(s, maxConcurrency, delayErrors);
         s.onSubscribe(parent);
-        source.subscribe(parent);
+        source.unsafeSubscribe(parent);
     }
 
     static final class CompletableMergeSubscriber

src/test/java/rx/internal/operators/CompletableConcatTest.java

@@ -0,0 +1,45 @@
+package rx.internal.operators;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.*;
+
+import rx.*;
+import rx.functions.*;
+import rx.schedulers.Schedulers;
+
+public class CompletableConcatTest {
+
+    @Test
+    public void asyncObservables() {
+
+        final int[] calls = { 0 };
+
+        Completable.concat(Observable.range(1, 5).map(new Func1<Integer, Completable>() {
+            @Override
+            public Completable call(final Integer v) {
+                System.out.println("Mapping " + v);
+                return Completable.fromAction(new Action0() {
+                    @Override
+                    public void call() {
+                        System.out.println("Processing " + (calls[0] + 1));
+                        calls[0]++;
+                    }
+                })
+                .subscribeOn(Schedulers.io())
+                .doOnCompleted(new Action0() {
+                    @Override
+                    public void call() {
+                        System.out.println("Inner complete " + v);
+                    }
+                })
+                .observeOn(Schedulers.computation());
+            }
+        })
+        ).test()
+        .awaitTerminalEventAndUnsubscribeOnTimeout(5, TimeUnit.SECONDS)
+        .assertResult();
+
+        Assert.assertEquals(5, calls[0]);
+    }
+}

src/test/java/rx/internal/operators/CompletableMergeTest.java

@@ -0,0 +1,45 @@
+package rx.internal.operators;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.*;
+
+import rx.*;
+import rx.functions.*;
+import rx.schedulers.Schedulers;
+
+public class CompletableMergeTest {
+
+    @Test
+    public void asyncObservables() {
+
+        final int[] calls = { 0 };
+
+        Completable.merge(Observable.range(1, 5).map(new Func1<Integer, Completable>() {
+            @Override
+            public Completable call(final Integer v) {
+                System.out.println("Mapping " + v);
+                return Completable.fromAction(new Action0() {
+                    @Override
+                    public void call() {
+                        System.out.println("Processing " + (calls[0] + 1));
+                        calls[0]++;
+                    }
+                })
+                .subscribeOn(Schedulers.io())
+                .doOnCompleted(new Action0() {
+                    @Override
+                    public void call() {
+                        System.out.println("Inner complete " + v);
+                    }
+                })
+                .observeOn(Schedulers.computation());
+            }
+        }), 1
+        ).test()
+        .awaitTerminalEventAndUnsubscribeOnTimeout(5, TimeUnit.SECONDS)
+        .assertResult();
+
+        Assert.assertEquals(5, calls[0]);
+    }
+}