2.x: fix wrong reference check in FlattenIterable (#4165)

Author: akarnokd

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

@@ -19,7 +19,8 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.functions.*;
+import io.reactivex.exceptions.MissingBackpressureException;
+import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.Objects;
 import io.reactivex.internal.fuseable.QueueSubscription;
 import io.reactivex.internal.queue.SpscArrayQueue;
@@ -166,7 +167,7 @@ public void onSubscribe(Subscription s) {
         @Override
         public void onNext(T t) {
             if (fusionMode != ASYNC && !queue.offer(t)) {
-                onError(new IllegalStateException("Queue is full?!"));
+                onError(new MissingBackpressureException("Queue is full?!"));
                 return;
             }
             drain();
@@ -363,13 +364,14 @@ boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, Queue<?> q) {
                 return true;
             }
             if (d) {
-                if (error != null) {
-                    Throwable e = Exceptions.terminate(error);
+                Throwable ex = error.get();
+                if (ex != null) {
+                    ex = Exceptions.terminate(error);
 
                     current = null;
                     q.clear();
 
-                    a.onError(e);
+                    a.onError(ex);
                     return true;
                 } else
                     if (empty) {

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

@@ -0,0 +1,50 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.flowable;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+
+import io.reactivex.Flowable;
+import io.reactivex.functions.*;
+import io.reactivex.subscribers.TestSubscriber;
+
+public class FlowableFlattenIterableTest {
+
+    @Test
+    public void normal() {
+        
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        
+        Flowable.range(1, 2)
+        .reduce(new BiFunction<Integer, Integer, Integer>() {
+            @Override
+            public Integer apply(Integer a, Integer b) {
+                return Math.max(a, b);
+            }
+        })
+        .flatMapIterable(new Function<Integer, Iterable<Integer>>() {
+            @Override
+            public Iterable<Integer> apply(Integer v) {
+                return Arrays.asList(v, v + 1);
+            }
+        })
+        .subscribe(ts);
+        
+        ts.assertValues(2, 3)
+        .assertNoErrors()
+        .assertComplete();
+    }
+}