2.x: Enable fusion-consumers (#4157)

* 2.x: Enable fusion-consumers

* Fix observeOn leaking the worker

* Code cleanup

Author: akarnokd

src/main/java/io/reactivex/Flowable.java

@@ -27,6 +27,7 @@
 import io.reactivex.internal.functions.Objects;
 import io.reactivex.internal.fuseable.ScalarCallable;
 import io.reactivex.internal.operators.flowable.*;
+import io.reactivex.internal.operators.flowable.FlowableConcatMap.ErrorMode;
 import io.reactivex.internal.subscribers.flowable.*;
 import io.reactivex.internal.subscriptions.EmptySubscription;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -1467,7 +1468,7 @@ public final <R> Flowable<R> concatMap(Function<? super T, ? extends Publisher<?
         if (prefetch <= 0) {
             throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);
         }
-        return new FlowableConcatMap<T, R>(this, mapper, prefetch);
+        return new FlowableConcatMap<T, R>(this, mapper, prefetch, ErrorMode.IMMEDIATE);
     }
 
     @BackpressureSupport(BackpressureKind.FULL)
@@ -1480,12 +1481,7 @@ public final <U> Flowable<U> concatMapIterable(Function<? super T, ? extends Ite
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> Flowable<U> concatMapIterable(final Function<? super T, ? extends Iterable<? extends U>> mapper, int prefetch) {
         Objects.requireNonNull(mapper, "mapper is null");
-        return concatMap(new Function<T, Publisher<U>>() {
-            @Override
-            public Publisher<U> apply(T v) {
-                return new FlowableFromIterable<U>(mapper.apply(v));
-            }
-        }, prefetch);
+        return new FlowableFlattenIterable<T, U>(this, mapper, prefetch);
     }
 
     @BackpressureSupport(BackpressureKind.FULL)
@@ -2029,13 +2025,7 @@ public final <U, R> Flowable<R> flatMap(Function<? super T, ? extends Publisher<
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> Flowable<U> flatMapIterable(final Function<? super T, ? extends Iterable<? extends U>> mapper) {
-        Objects.requireNonNull(mapper, "mapper is null");
-        return flatMap(new Function<T, Publisher<U>>() {
-            @Override
-            public Publisher<U> apply(T v) {
-                return new FlowableFromIterable<U>(mapper.apply(v));
-            }
-        });
+        return flatMapIterable(mapper, bufferSize());
     }
 
     @BackpressureSupport(BackpressureKind.FULL)
@@ -2054,12 +2044,7 @@ public Publisher<U> apply(T t) {
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> Flowable<U> flatMapIterable(final Function<? super T, ? extends Iterable<? extends U>> mapper, int bufferSize) {
-        return flatMap(new Function<T, Publisher<U>>() {
-            @Override
-            public Publisher<U> apply(T v) {
-                return new FlowableFromIterable<U>(mapper.apply(v));
-            }
-        }, false, bufferSize);
+        return new FlowableFlattenIterable<T, U>(this, mapper, bufferSize);
     }
 
     @BackpressureSupport(BackpressureKind.NONE)

src/main/java/io/reactivex/internal/operators/completable/CompletableConcat.java

@@ -68,7 +68,7 @@ public CompletableConcatSubscriber(CompletableSubscriber actual, int prefetch) {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
                 actual.onSubscribe(this);
                 s.request(prefetch);

src/main/java/io/reactivex/internal/operators/completable/CompletableMerge.java

@@ -82,7 +82,7 @@ public boolean isDisposed() {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
                 set.add(Disposables.from(s));
                 actual.onSubscribe(this);

src/main/java/io/reactivex/internal/operators/flowable/FlowableAll.java

@@ -50,7 +50,7 @@ public AllSubscriber(Subscriber<? super Boolean> actual, Predicate<? super T> pr
         }
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
                 actual.onSubscribe(this);
                 s.request(Long.MAX_VALUE);

src/main/java/io/reactivex/internal/operators/flowable/FlowableAmb.java

@@ -95,7 +95,7 @@ public void subscribe(Publisher<? extends T>[] sources) {
 
         @Override
         public void request(long n) {
-            if (!SubscriptionHelper.validateRequest(n)) {
+            if (!SubscriptionHelper.validate(n)) {
                 return;
             }
 
@@ -172,7 +172,7 @@ public void request(long n) {
             Subscription s = get();
             if (s != null) {
                 s.request(n);
-            } else if (SubscriptionHelper.validateRequest(n)) {
+            } else if (SubscriptionHelper.validate(n)) {
                 BackpressureHelper.add(missedRequested, n);
                 s = get();
                 if (s != null && s != SubscriptionHelper.CANCELLED) {

src/main/java/io/reactivex/internal/operators/flowable/FlowableAny.java

@@ -47,7 +47,7 @@ public AnySubscriber(Subscriber<? super Boolean> actual, Predicate<? super T> pr
         }
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
                 actual.onSubscribe(this);
                 s.request(Long.MAX_VALUE);

src/main/java/io/reactivex/internal/operators/flowable/FlowableBuffer.java

@@ -96,7 +96,7 @@ boolean createBuffer() {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
                 actual.onSubscribe(this);
             }
@@ -137,7 +137,7 @@ public void onComplete() {
 
         @Override
         public void request(long n) {
-            if (SubscriptionHelper.validateRequest(n)) {
+            if (SubscriptionHelper.validate(n)) {
                 long m = BackpressureHelper.multiplyCap(n, count);
                 s.request(m);
             }
@@ -173,7 +173,7 @@ public BufferSkipSubscriber(Subscriber<? super U> actual, int count, int skip, S
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
                 actual.onSubscribe(this);
             }
@@ -231,7 +231,7 @@ public void onComplete() {
 
         @Override
         public void request(long n) {
-            if (!SubscriptionHelper.validateRequest(n)) {
+            if (!SubscriptionHelper.validate(n)) {
                 return;
             }
             // requesting the first set of buffers must happen only once

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundary.java

@@ -77,7 +77,7 @@ public BufferBoundarySubscriber(Subscriber<? super U> actual,
         }
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (SubscriptionHelper.validate(this.s, s)) {
                 this.s = s;
 
                 BufferOpenSubscriber<T, U, Open, Close> bos = new BufferOpenSubscriber<T, U, Open, Close>(this);

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundarySupplier.java

@@ -67,7 +67,7 @@ public BufferBondarySupplierSubscriber(Subscriber<? super U> actual, Supplier<U>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (!SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (!SubscriptionHelper.validate(this.s, s)) {
                 return;
             }
             this.s = s;

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferExactBoundary.java

@@ -64,7 +64,7 @@ public BufferExactBondarySubscriber(Subscriber<? super U> actual, Supplier<U> bu
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (!SubscriptionHelper.validateSubscription(this.s, s)) {
+            if (!SubscriptionHelper.validate(this.s, s)) {
                 return;
             }
             this.s = s;