2.x: fix Observable.concatMapEager bad logic for immediate scalars (#4982)

* 2.x: fix Observable.concatMapEager bad logic for immediate scalars

* Don't print-log

Author: akarnokd

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

@@ -14,7 +14,6 @@
 package io.reactivex.internal.operators.observable;
 
 import java.util.ArrayDeque;
-import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import io.reactivex.*;
@@ -218,7 +217,6 @@ public void innerComplete(InnerQueuedObserver<R> inner) {
             drain();
         }
 
-        @SuppressWarnings("unchecked")
         @Override
         public void drain() {
             if (getAndIncrement() != 0) {
@@ -276,23 +274,6 @@ public void drain() {
                         return;
                     }
 
-                    if (source instanceof Callable) {
-                        R w;
-
-                        try {
-                            w = ((Callable<R>)source).call();
-                        } catch (Throwable ex) {
-                            Exceptions.throwIfFatal(ex);
-                            error.addThrowable(ex);
-                            continue;
-                        }
-
-                        if (w != null) {
-                            a.onNext(w);
-                        }
-                        continue;
-                    }
-
                     InnerQueuedObserver<R> inner = new InnerQueuedObserver<R>(this, prefetch);
 
                     observers.offer(inner);

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

@@ -1175,4 +1175,22 @@ public void innerLong() {
         .assertComplete()
         .assertNoErrors();
     }
+
+    @Test
+    public void oneDelayed() {
+        Flowable.just(1, 2, 3, 4, 5)
+        .concatMapEager(new Function<Integer, Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(Integer i) throws Exception {
+                return i == 3 ? Flowable.just(i) : Flowable
+                        .just(i)
+                        .delay(1, TimeUnit.MILLISECONDS, Schedulers.io());
+            }
+        })
+        .observeOn(Schedulers.io())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(1, 2, 3, 4, 5)
+        ;
+    }
 }

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

@@ -983,4 +983,22 @@ public ObservableSource<Object> apply(Object v) throws Exception {
             }
         });
     }
+
+    @Test
+    public void oneDelayed() {
+        Observable.just(1, 2, 3, 4, 5)
+        .concatMapEager(new Function<Integer, ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> apply(Integer i) throws Exception {
+                return i == 3 ? Observable.just(i) : Observable
+                        .just(i)
+                        .delay(1, TimeUnit.MILLISECONDS, Schedulers.io());
+            }
+        })
+        .observeOn(Schedulers.io())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(1, 2, 3, 4, 5)
+        ;
+    }
 }