2.x fix Flowable.create() not reporting null values properly, unify (#4583)

akarnokd · web-flow · commit 24448b47ed27 · 2016-09-22T21:07:56.000+02:00
exception msg across
diff --git a/src/main/java/io/reactivex/internal/operators/completable/CompletableCreate.java b/src/main/java/io/reactivex/internal/operators/completable/CompletableCreate.java
@@ -74,7 +74,7 @@ public void onComplete() {
         @Override
         public void onError(Throwable t) {
             if (t == null) {
-                t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
             }
             if (get() != DisposableHelper.DISPOSED) {
                 Disposable d = getAndSet(DisposableHelper.DISPOSED);
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableCreate.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableCreate.java
@@ -107,7 +107,7 @@ public void onNext(T t) {
                 return;
             }
             if (t == null) {
-                onError(new NullPointerException("Emitter got a null value. Null values are generally not allowed in 2.x operators and sources."));
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
                 return;
             }
             if (get() == 0 && compareAndSet(0, 1)) {
@@ -134,7 +134,7 @@ public void onError(Throwable t) {
                 return;
             }
             if (t == null) {
-                t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
             }
             if (error.addThrowable(t)) {
                 done = true;
@@ -265,6 +265,7 @@ public void onComplete() {
         @Override
         public void onError(Throwable e) {
             if (isCancelled()) {
+                RxJavaPlugins.onError(e);
                 return;
             }
             try {
@@ -337,7 +338,12 @@ public void onNext(T t) {
                 return;
             }
 
-            actual.onNext(t);
+            if (t != null) {
+                actual.onNext(t);
+            } else {
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
+            }
 
             for (;;) {
                 long r = get();
@@ -363,6 +369,11 @@ public final void onNext(T t) {
                 return;
             }
 
+            if (t == null) {
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
+            }
+
             if (get() != 0) {
                 actual.onNext(t);
                 BackpressureHelper.produced(this, 1);
@@ -426,12 +437,29 @@ static final class BufferAsyncEmitter<T> extends BaseEmitter<T> {
 
         @Override
         public void onNext(T t) {
+            if (done || isCancelled()) {
+                return;
+            }
+
+            if (t == null) {
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
+            }
             queue.offer(t);
             drain();
         }
 
         @Override
         public void onError(Throwable e) {
+            if (done || isCancelled()) {
+                RxJavaPlugins.onError(e);
+                return;
+            }
+
+            if (e == null) {
+                e = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+            }
+
             error = e;
             done = true;
             drain();
@@ -552,12 +580,27 @@ static final class LatestAsyncEmitter<T> extends BaseEmitter<T> {
 
         @Override
         public void onNext(T t) {
+            if (done || isCancelled()) {
+                return;
+            }
+
+            if (t == null) {
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
+            }
             queue.set(t);
             drain();
         }
 
         @Override
         public void onError(Throwable e) {
+            if (done || isCancelled()) {
+                RxJavaPlugins.onError(e);
+                return;
+            }
+            if (e == null) {
+                onError(new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources."));
+            }
             error = e;
             done = true;
             drain();
diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeCreate.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeCreate.java
@@ -69,7 +69,7 @@ public void onSuccess(T value) {
                 if (d != DisposableHelper.DISPOSED) {
                     try {
                         if (value == null) {
-                            actual.onError(new NullPointerException("Emitter got a null value. Null values are generally not allowed in 2.x operators and sources."));
+                            actual.onError(new NullPointerException("onSuccess called with null. Null values are generally not allowed in 2.x operators and sources."));
                         } else {
                             actual.onSuccess(value);
                         }
@@ -85,7 +85,7 @@ public void onSuccess(T value) {
         @Override
         public void onError(Throwable t) {
             if (t == null) {
-                t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
             }
             if (get() != DisposableHelper.DISPOSED) {
                 Disposable d = getAndSet(DisposableHelper.DISPOSED);
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableCreate.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableCreate.java
@@ -60,7 +60,8 @@ static final class CreateEmitter<T>
         @Override
         public void onNext(T t) {
             if (t == null) {
-                onError(new NullPointerException("Emitter got a null value. Null values are generally not allowed in 2.x operators and sources."));
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+                return;
             }
             if (!isDisposed()) {
                 observer.onNext(t);
@@ -70,7 +71,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (t == null) {
-                t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
             }
             if (!isDisposed()) {
                 try {
@@ -151,7 +152,7 @@ public void onNext(T t) {
                 return;
             }
             if (t == null) {
-                onError(new NullPointerException("t is null"));
+                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
                 return;
             }
             if (get() == 0 && compareAndSet(0, 1)) {
@@ -178,7 +179,7 @@ public void onError(Throwable t) {
                 return;
             }
             if (t == null) {
-                t = new NullPointerException("t is null");
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
             }
             if (error.addThrowable(t)) {
                 done = true;
diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleCreate.java b/src/main/java/io/reactivex/internal/operators/single/SingleCreate.java
@@ -63,7 +63,7 @@ public void onSuccess(T value) {
                 if (d != DisposableHelper.DISPOSED) {
                     try {
                         if (value == null) {
-                            actual.onError(new NullPointerException("Emitter got a null value. Null values are generally not allowed in 2.x operators and sources."));
+                            actual.onError(new NullPointerException("onSuccess called with null. Null values are generally not allowed in 2.x operators and sources."));
                         } else {
                             actual.onSuccess(value);
                         }
@@ -79,7 +79,7 @@ public void onSuccess(T value) {
         @Override
         public void onError(Throwable t) {
             if (t == null) {
-                t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");
+                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
             }
             if (get() != DisposableHelper.DISPOSED) {
                 Disposable d = getAndSet(DisposableHelper.DISPOSED);
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableCreateTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableCreateTest.java
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableCreateTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableCreateTest.java

Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ public void onComplete() {`
`74`	`74`	`@Override`
`75`	`75`	`public void onError(Throwable t) {`
`76`	`76`	`if (t == null) {`
`77`		`- t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");`
	`77`	`+ t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");`
`78`	`78`	`}`
`79`	`79`	`if (get() != DisposableHelper.DISPOSED) {`
`80`	`80`	`Disposable d = getAndSet(DisposableHelper.DISPOSED);`
Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,8 @@ static final class CreateEmitter<T>`
`60`	`60`	`@Override`
`61`	`61`	`public void onNext(T t) {`
`62`	`62`	`if (t == null) {`
`63`		`- onError(new NullPointerException("Emitter got a null value. Null values are generally not allowed in 2.x operators and sources."));`
	`63`	`+ onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));`
	`64`	`+ return;`
`64`	`65`	`}`
`65`	`66`	`if (!isDisposed()) {`
`66`	`67`	`observer.onNext(t);`
`@@ -70,7 +71,7 @@ public void onNext(T t) {`
`70`	`71`	`@Override`
`71`	`72`	`public void onError(Throwable t) {`
`72`	`73`	`if (t == null) {`
`73`		`- t = new NullPointerException("Emitter got a null throwable. Null values are generally not allowed in 2.x operators and sources.");`
	`74`	`+ t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");`
`74`	`75`	`}`
`75`	`76`	`if (!isDisposed()) {`
`76`	`77`	`try {`
`@@ -151,7 +152,7 @@ public void onNext(T t) {`
`151`	`152`	`return;`
`152`	`153`	`}`
`153`	`154`	`if (t == null) {`
`154`		`- onError(new NullPointerException("t is null"));`
	`155`	`+ onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));`
`155`	`156`	`return;`
`156`	`157`	`}`
`157`	`158`	`if (get() == 0 && compareAndSet(0, 1)) {`
`@@ -178,7 +179,7 @@ public void onError(Throwable t) {`
`178`	`179`	`return;`
`179`	`180`	`}`
`180`	`181`	`if (t == null) {`
`181`		`- t = new NullPointerException("t is null");`
	`182`	`+ t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");`
`182`	`183`	`}`
`183`	`184`	`if (error.addThrowable(t)) {`
`184`	`185`	`done = true;`