Merge branch '2.x' of https://github.com/ReactiveX/RxJava.git into 2.x

Author: akarnokd

src/main/java/io/reactivex/internal/operators/single/SingleFlatMap.java

@@ -17,7 +17,9 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Function;
-import io.reactivex.internal.disposables.SequentialDisposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.functions.ObjectHelper;
+import java.util.concurrent.atomic.AtomicReference;
 
 public final class SingleFlatMap<T, R> extends Single<R> {
     final SingleSource<? extends T> source;
@@ -30,72 +32,84 @@ public SingleFlatMap(SingleSource<? extends T> source, Function<? super T, ? ext
     }
 
     @Override
-    protected void subscribeActual(SingleObserver<? super R> subscriber) {
-        SingleFlatMapCallback<T, R> parent = new SingleFlatMapCallback<T, R>(subscriber, mapper);
-        subscriber.onSubscribe(parent.sd);
-        source.subscribe(parent);
+    protected void subscribeActual(SingleObserver<? super R> actual) {
+        source.subscribe(new SingleFlatMapCallback<T, R>(actual, mapper));
     }
 
-    static final class SingleFlatMapCallback<T, R> implements SingleObserver<T> {
+    static final class SingleFlatMapCallback<T, R>
+    extends AtomicReference<Disposable>
+    implements SingleObserver<T>, Disposable {
         final SingleObserver<? super R> actual;
         final Function<? super T, ? extends SingleSource<? extends R>> mapper;
 
-        final SequentialDisposable sd;
-
         SingleFlatMapCallback(SingleObserver<? super R> actual,
                 Function<? super T, ? extends SingleSource<? extends R>> mapper) {
             this.actual = actual;
             this.mapper = mapper;
-            this.sd = new SequentialDisposable();
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
         }
 
         @Override
         public void onSubscribe(Disposable d) {
-            sd.replace(d);
+            if (DisposableHelper.setOnce(this, d)) {
+                actual.onSubscribe(this);
+            }
         }
 
         @Override
         public void onSuccess(T value) {
             SingleSource<? extends R> o;
 
             try {
-                o = mapper.apply(value);
+                o = ObjectHelper.requireNonNull(mapper.apply(value), "The single returned by the mapper is null");
             } catch (Throwable e) {
                 Exceptions.throwIfFatal(e);
                 actual.onError(e);
                 return;
             }
 
-            if (o == null) {
-                actual.onError(new NullPointerException("The single returned by the mapper is null"));
-                return;
-            }
-
-            if (sd.isDisposed()) {
-                return;
-            }
-
-            o.subscribe(new SingleObserver<R>() {
-                @Override
-                public void onSubscribe(Disposable d) {
-                    sd.replace(d);
-                }
-
-                @Override
-                public void onSuccess(R value) {
-                    actual.onSuccess(value);
-                }
-
-                @Override
-                public void onError(Throwable e) {
-                    actual.onError(e);
-                }
-            });
+            o.subscribe(new FlatMapSingleObserver<R>(this, actual));
         }
 
         @Override
         public void onError(Throwable e) {
             actual.onError(e);
         }
+
+        static final class FlatMapSingleObserver<R> implements SingleObserver<R> {
+
+            final AtomicReference<Disposable> parent;
+
+            final SingleObserver<? super R> actual;
+
+            FlatMapSingleObserver(AtomicReference<Disposable> parent, SingleObserver<? super R> actual) {
+                this.parent = parent;
+                this.actual = actual;
+            }
+
+            @Override
+            public void onSubscribe(final Disposable d) {
+                DisposableHelper.replace(parent, d);
+            }
+
+            @Override
+            public void onSuccess(final R value) {
+                actual.onSuccess(value);
+            }
+
+            @Override
+            public void onError(final Throwable e) {
+                actual.onError(e);
+            }
+        }
     }
 }

src/test/java/io/reactivex/internal/operators/single/SingleFlatMapTest.java

@@ -126,4 +126,79 @@ public Publisher<Integer> apply(Integer v) throws Exception {
         .test()
         .assertResult(1, 2, 3, 4, 5);
     }
+
+    @Test(expected = NullPointerException.class)
+    public void flatMapNull() {
+        Single.just(1)
+            .flatMap(null);
+    }
+
+    @Test
+    public void flatMapValue() {
+        Single.just(1).flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override public SingleSource<Integer> apply(final Integer integer) throws Exception {
+                if (integer == 1) {
+                    return Single.just(2);
+                }
+
+                return Single.just(1);
+            }
+        })
+            .test()
+            .assertResult(2);
+    }
+
+    @Test
+    public void flatMapValueDifferentType() {
+        Single.just(1).flatMap(new Function<Integer, SingleSource<String>>() {
+            @Override public SingleSource<String> apply(final Integer integer) throws Exception {
+                if (integer == 1) {
+                    return Single.just("2");
+                }
+
+                return Single.just("1");
+            }
+        })
+            .test()
+            .assertResult("2");
+    }
+
+    @Test
+    public void flatMapValueNull() {
+        Single.just(1).flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override public SingleSource<Integer> apply(final Integer integer) throws Exception {
+                return null;
+            }
+        })
+            .test()
+            .assertNoValues()
+            .assertError(NullPointerException.class)
+            .assertErrorMessage("The single returned by the mapper is null");
+    }
+
+    @Test
+    public void flatMapValueErrorThrown() {
+        Single.just(1).flatMap(new Function<Integer, SingleSource<Integer>>() {
+            @Override public SingleSource<Integer> apply(final Integer integer) throws Exception {
+                throw new RuntimeException("something went terribly wrong!");
+            }
+        })
+            .test()
+            .assertNoValues()
+            .assertError(RuntimeException.class)
+            .assertErrorMessage("something went terribly wrong!");
+    }
+
+    @Test
+    public void flatMapError() {
+        RuntimeException exception = new RuntimeException("test");
+
+        Single.error(exception).flatMap(new Function<Object, SingleSource<Object>>() {
+            @Override public SingleSource<Object> apply(final Object integer) throws Exception {
+                return Single.just(new Object());
+            }
+        })
+            .test()
+            .assertError(exception);
+    }
 }