2.x: Fix Observable.concatMapEager queueing of source items (#5609)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMapEager.java

@@ -24,6 +24,7 @@
 import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.observers.*;
+import io.reactivex.internal.queue.SpscLinkedArrayQueue;
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
@@ -129,7 +130,7 @@ public void onSubscribe(Disposable d) {
                     }
                 }
 
-                queue = QueueDrainHelper.createQueue(prefetch);
+                queue = new SpscLinkedArrayQueue<T>(prefetch);
 
                 actual.onSubscribe(this);
             }

src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapEagerTest.java

@@ -1193,4 +1193,39 @@ public Flowable<Integer> apply(Integer i) throws Exception {
         .assertResult(1, 2, 3, 4, 5)
         ;
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void maxConcurrencyOf2() {
+        List<Integer>[] list = new ArrayList[100];
+        for (int i = 0; i < 100; i++) {
+            List<Integer> lst = new ArrayList<Integer>();
+            list[i] = lst;
+            for (int k = 1; k <= 10; k++) {
+                lst.add((i) * 10 + k);
+            }
+        }
+
+        Flowable.range(1, 1000)
+        .buffer(10)
+        .concatMapEager(new Function<List<Integer>, Flowable<List<Integer>>>() {
+            @Override
+            public Flowable<List<Integer>> apply(List<Integer> v)
+                    throws Exception {
+                return Flowable.just(v)
+                        .subscribeOn(Schedulers.io())
+                        .doOnNext(new Consumer<List<Integer>>() {
+                            @Override
+                            public void accept(List<Integer> v)
+                                    throws Exception {
+                                Thread.sleep(new Random().nextInt(20));
+                            }
+                        });
+            }
+        }
+                , 2, 3)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(list);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapEagerTest.java

@@ -1001,4 +1001,39 @@ public ObservableSource<Integer> apply(Integer i) throws Exception {
         .assertResult(1, 2, 3, 4, 5)
         ;
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void maxConcurrencyOf2() {
+        List<Integer>[] list = new ArrayList[100];
+        for (int i = 0; i < 100; i++) {
+            List<Integer> lst = new ArrayList<Integer>();
+            list[i] = lst;
+            for (int k = 1; k <= 10; k++) {
+                lst.add((i) * 10 + k);
+            }
+        }
+
+        Observable.range(1, 1000)
+        .buffer(10)
+        .concatMapEager(new Function<List<Integer>, ObservableSource<List<Integer>>>() {
+            @Override
+            public ObservableSource<List<Integer>> apply(List<Integer> v)
+                    throws Exception {
+                return Observable.just(v)
+                        .subscribeOn(Schedulers.io())
+                        .doOnNext(new Consumer<List<Integer>>() {
+                            @Override
+                            public void accept(List<Integer> v)
+                                    throws Exception {
+                                Thread.sleep(new Random().nextInt(20));
+                            }
+                        });
+            }
+        }
+                , 2, 3)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(list);
+    }
 }