2.x: Subject NPE fixes, add UnicastProcessor TCK (#5760)

* 2.x: add Subject/Processor refCount(), Subject NPE fixes

* Fix wording

* Move RefCountProcessor into tests

* Improve style

Author: akarnokd

src/main/java/io/reactivex/processors/AsyncProcessor.java

@@ -15,10 +15,12 @@
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.reactivestreams.*;
+
 import io.reactivex.annotations.*;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.subscriptions.DeferredScalarSubscription;
 import io.reactivex.plugins.RxJavaPlugins;
-import org.reactivestreams.*;
 
 /**
  * Processor that emits the very last value followed by a completion event or the received error
@@ -77,32 +79,17 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(T t) {
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (subscribers.get() == TERMINATED) {
             return;
         }
-        if (t == null) {
-            nullOnNext();
-            return;
-        }
         value = t;
     }
 
-    @SuppressWarnings("unchecked")
-    void nullOnNext() {
-        value = null;
-        Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-        error = ex;
-        for (AsyncSubscription<T> as : subscribers.getAndSet(TERMINATED)) {
-            as.onError(ex);
-        }
-    }
-
     @SuppressWarnings("unchecked")
     @Override
     public void onError(Throwable t) {
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (subscribers.get() == TERMINATED) {
             RxJavaPlugins.onError(t);
             return;

src/main/java/io/reactivex/processors/BehaviorProcessor.java

@@ -175,10 +175,8 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(T t) {
-        if (t == null) {
-            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
-            return;
-        }
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
+
         if (terminalEvent.get() != null) {
             return;
         }
@@ -191,9 +189,7 @@ public void onNext(T t) {
 
     @Override
     public void onError(Throwable t) {
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (!terminalEvent.compareAndSet(null, t)) {
             RxJavaPlugins.onError(t);
             return;

src/main/java/io/reactivex/processors/FlowableProcessor.java

@@ -13,10 +13,11 @@
 
 package io.reactivex.processors;
 
-import io.reactivex.*;
-import io.reactivex.annotations.NonNull;
 import org.reactivestreams.Processor;
 
+import io.reactivex.*;
+import io.reactivex.annotations.*;
+
 /**
  * Represents a Subscriber and a Flowable (Publisher) at the same time, allowing
  * multicasting events from a single source to multiple child Subscribers.
@@ -28,45 +29,47 @@
 public abstract class FlowableProcessor<T> extends Flowable<T> implements Processor<T, T>, FlowableSubscriber<T> {
 
     /**
-     * Returns true if the subject has subscribers.
+     * Returns true if the FlowableProcessor has subscribers.
      * <p>The method is thread-safe.
-     * @return true if the subject has subscribers
+     * @return true if the FlowableProcessor has subscribers
      */
     public abstract boolean hasSubscribers();
 
     /**
-     * Returns true if the subject has reached a terminal state through an error event.
+     * Returns true if the FlowableProcessor has reached a terminal state through an error event.
      * <p>The method is thread-safe.
-     * @return true if the subject has reached a terminal state through an error event
+     * @return true if the FlowableProcessor has reached a terminal state through an error event
      * @see #getThrowable()
      * @see #hasComplete()
      */
     public abstract boolean hasThrowable();
 
     /**
-     * Returns true if the subject has reached a terminal state through a complete event.
+     * Returns true if the FlowableProcessor has reached a terminal state through a complete event.
      * <p>The method is thread-safe.
-     * @return true if the subject has reached a terminal state through a complete event
+     * @return true if the FlowableProcessor has reached a terminal state through a complete event
      * @see #hasThrowable()
      */
     public abstract boolean hasComplete();
 
     /**
-     * Returns the error that caused the Subject to terminate or null if the Subject
+     * Returns the error that caused the FlowableProcessor to terminate or null if the FlowableProcessor
      * hasn't terminated yet.
      * <p>The method is thread-safe.
-     * @return the error that caused the Subject to terminate or null if the Subject
+     * @return the error that caused the FlowableProcessor to terminate or null if the FlowableProcessor
      * hasn't terminated yet
      */
+    @Nullable
     public abstract Throwable getThrowable();
 
     /**
-     * Wraps this Subject and serializes the calls to the onSubscribe, onNext, onError and
+     * Wraps this FlowableProcessor and serializes the calls to the onSubscribe, onNext, onError and
      * onComplete methods, making them thread-safe.
      * <p>The method is thread-safe.
-     * @return the wrapped and serialized subject
+     * @return the wrapped and serialized FlowableProcessor
      */
     @NonNull
+    @CheckReturnValue
     public final FlowableProcessor<T> toSerialized() {
         if (this instanceof SerializedProcessor) {
             return this;

src/main/java/io/reactivex/processors/PublishProcessor.java

@@ -18,6 +18,7 @@
 
 import io.reactivex.annotations.*;
 import io.reactivex.exceptions.MissingBackpressureException;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -186,13 +187,10 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(T t) {
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (subscribers.get() == TERMINATED) {
             return;
         }
-        if (t == null) {
-            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
-            return;
-        }
         for (PublishSubscription<T> s : subscribers.get()) {
             s.onNext(t);
         }
@@ -201,13 +199,11 @@ public void onNext(T t) {
     @SuppressWarnings("unchecked")
     @Override
     public void onError(Throwable t) {
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (subscribers.get() == TERMINATED) {
             RxJavaPlugins.onError(t);
             return;
         }
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
         error = t;
 
         for (PublishSubscription<T> s : subscribers.getAndSet(TERMINATED)) {

src/main/java/io/reactivex/processors/ReplayProcessor.java

@@ -267,10 +267,8 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(T t) {
-        if (t == null) {
-            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
-            return;
-        }
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
+
         if (done) {
             return;
         }
@@ -286,9 +284,8 @@ public void onNext(T t) {
     @SuppressWarnings("unchecked")
     @Override
     public void onError(Throwable t) {
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+
         if (done) {
             RxJavaPlugins.onError(t);
             return;

src/main/java/io/reactivex/processors/UnicastProcessor.java

@@ -339,12 +339,9 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(T t) {
-        if (done || cancelled) {
-            return;
-        }
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
 
-        if (t == null) {
-            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
+        if (done || cancelled) {
             return;
         }
 
@@ -354,15 +351,13 @@ public void onNext(T t) {
 
     @Override
     public void onError(Throwable t) {
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+
         if (done || cancelled) {
             RxJavaPlugins.onError(t);
             return;
         }
 
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
-
         error = t;
         done = true;
 

src/main/java/io/reactivex/subjects/AsyncSubject.java

@@ -13,12 +13,13 @@
 
 package io.reactivex.subjects;
 
-import io.reactivex.annotations.CheckReturnValue;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.Observer;
+import io.reactivex.annotations.CheckReturnValue;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.observers.DeferredScalarDisposable;
 import io.reactivex.plugins.RxJavaPlugins;
 
@@ -75,32 +76,17 @@ public void onSubscribe(Disposable s) {
 
     @Override
     public void onNext(T t) {
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (subscribers.get() == TERMINATED) {
             return;
         }
-        if (t == null) {
-            nullOnNext();
-            return;
-        }
         value = t;
     }
 
-    @SuppressWarnings("unchecked")
-    void nullOnNext() {
-        value = null;
-        Throwable ex = new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-        error = ex;
-        for (AsyncDisposable<T> as : subscribers.getAndSet(TERMINATED)) {
-            as.onError(ex);
-        }
-    }
-
     @SuppressWarnings("unchecked")
     @Override
     public void onError(Throwable t) {
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (subscribers.get() == TERMINATED) {
             RxJavaPlugins.onError(t);
             return;

src/main/java/io/reactivex/subjects/BehaviorSubject.java

@@ -171,10 +171,8 @@ public void onSubscribe(Disposable s) {
 
     @Override
     public void onNext(T t) {
-        if (t == null) {
-            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
-            return;
-        }
+        ObjectHelper.requireNonNull(t, "onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
+
         if (terminalEvent.get() != null) {
             return;
         }
@@ -187,9 +185,7 @@ public void onNext(T t) {
 
     @Override
     public void onError(Throwable t) {
-        if (t == null) {
-            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-        }
+        ObjectHelper.requireNonNull(t, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (!terminalEvent.compareAndSet(null, t)) {
             RxJavaPlugins.onError(t);
             return;

src/main/java/io/reactivex/subjects/CompletableSubject.java

@@ -16,8 +16,9 @@
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.*;
-import io.reactivex.annotations.*;
+import io.reactivex.annotations.CheckReturnValue;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -66,9 +67,7 @@ public void onSubscribe(Disposable d) {
 
     @Override
     public void onError(Throwable e) {
-        if (e == null) {
-            e = new NullPointerException("Null errors are not allowed in 2.x");
-        }
+        ObjectHelper.requireNonNull(e, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (once.compareAndSet(false, true)) {
             this.error = e;
             for (CompletableDisposable md : observers.getAndSet(TERMINATED)) {

src/main/java/io/reactivex/subjects/MaybeSubject.java

@@ -18,6 +18,7 @@
 import io.reactivex.*;
 import io.reactivex.annotations.*;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -73,10 +74,7 @@ public void onSubscribe(Disposable d) {
     @SuppressWarnings("unchecked")
     @Override
     public void onSuccess(T value) {
-        if (value == null) {
-            onError(new NullPointerException("Null values are not allowed in 2.x"));
-            return;
-        }
+        ObjectHelper.requireNonNull(value, "onSuccess called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (once.compareAndSet(false, true)) {
             this.value = value;
             for (MaybeDisposable<T> md : observers.getAndSet(TERMINATED)) {
@@ -88,9 +86,7 @@ public void onSuccess(T value) {
     @SuppressWarnings("unchecked")
     @Override
     public void onError(Throwable e) {
-        if (e == null) {
-            e = new NullPointerException("Null errors are not allowed in 2.x");
-        }
+        ObjectHelper.requireNonNull(e, "onError called with null. Null values are generally not allowed in 2.x operators and sources.");
         if (once.compareAndSet(false, true)) {
             this.error = e;
             for (MaybeDisposable<T> md : observers.getAndSet(TERMINATED)) {