Add @nonnull annotations to create methods of Subjects and Processors (#5930)

Author: vpriscan

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

@@ -189,6 +189,7 @@ public final class BehaviorProcessor<T> extends FlowableProcessor<T> {
      * @return the constructed {@link BehaviorProcessor}
      */
     @CheckReturnValue
+    @NonNull
     public static <T> BehaviorProcessor<T> create() {
         return new BehaviorProcessor<T>();
     }
@@ -205,6 +206,7 @@ public static <T> BehaviorProcessor<T> create() {
      * @return the constructed {@link BehaviorProcessor}
      */
     @CheckReturnValue
+    @NonNull
     public static <T> BehaviorProcessor<T> createDefault(T defaultValue) {
         ObjectHelper.requireNonNull(defaultValue, "defaultValue is null");
         return new BehaviorProcessor<T>(defaultValue);

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

@@ -76,6 +76,7 @@ public final class PublishProcessor<T> extends FlowableProcessor<T> {
      * @return the new PublishProcessor
      */
     @CheckReturnValue
+    @NonNull
     public static <T> PublishProcessor<T> create() {
         return new PublishProcessor<T>();
     }

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

@@ -117,6 +117,7 @@ public final class ReplayProcessor<T> extends FlowableProcessor<T> {
      * @return the created ReplayProcessor
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplayProcessor<T> create() {
         return new ReplayProcessor<T>(new UnboundedReplayBuffer<T>(16));
     }
@@ -137,6 +138,7 @@ public static <T> ReplayProcessor<T> create() {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplayProcessor<T> create(int capacityHint) {
         return new ReplayProcessor<T>(new UnboundedReplayBuffer<T>(capacityHint));
     }
@@ -162,6 +164,7 @@ public static <T> ReplayProcessor<T> create(int capacityHint) {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplayProcessor<T> createWithSize(int maxSize) {
         return new ReplayProcessor<T>(new SizeBoundReplayBuffer<T>(maxSize));
     }
@@ -216,6 +219,7 @@ public static <T> ReplayProcessor<T> createWithSize(int maxSize) {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplayProcessor<T> createWithTime(long maxAge, TimeUnit unit, Scheduler scheduler) {
         return new ReplayProcessor<T>(new SizeAndTimeBoundReplayBuffer<T>(Integer.MAX_VALUE, maxAge, unit, scheduler));
     }
@@ -255,6 +259,7 @@ public static <T> ReplayProcessor<T> createWithTime(long maxAge, TimeUnit unit,
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplayProcessor<T> createWithTimeAndSize(long maxAge, TimeUnit unit, Scheduler scheduler, int maxSize) {
         return new ReplayProcessor<T>(new SizeAndTimeBoundReplayBuffer<T>(maxSize, maxAge, unit, scheduler));
     }

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

@@ -18,6 +18,7 @@
 
 import io.reactivex.annotations.Experimental;
 import io.reactivex.annotations.Nullable;
+import io.reactivex.annotations.NonNull;
 import org.reactivestreams.*;
 
 import io.reactivex.internal.functions.ObjectHelper;
@@ -74,6 +75,7 @@ public final class UnicastProcessor<T> extends FlowableProcessor<T> {
      * @return an UnicastSubject instance
      */
     @CheckReturnValue
+    @NonNull
     public static <T> UnicastProcessor<T> create() {
         return new UnicastProcessor<T>(bufferSize());
     }
@@ -85,6 +87,7 @@ public static <T> UnicastProcessor<T> create() {
      * @return an UnicastProcessor instance
      */
     @CheckReturnValue
+    @NonNull
     public static <T> UnicastProcessor<T> create(int capacityHint) {
         return new UnicastProcessor<T>(capacityHint);
     }
@@ -98,6 +101,7 @@ public static <T> UnicastProcessor<T> create(int capacityHint) {
      */
     @CheckReturnValue
     @Experimental
+    @NonNull
     public static <T> UnicastProcessor<T> create(boolean delayError) {
         return new UnicastProcessor<T>(bufferSize(), null, delayError);
     }
@@ -115,6 +119,7 @@ public static <T> UnicastProcessor<T> create(boolean delayError) {
      * @return an UnicastProcessor instance
      */
     @CheckReturnValue
+    @NonNull
     public static <T> UnicastProcessor<T> create(int capacityHint, Runnable onCancelled) {
         ObjectHelper.requireNonNull(onCancelled, "onTerminate");
         return new UnicastProcessor<T>(capacityHint, onCancelled);
@@ -136,6 +141,7 @@ public static <T> UnicastProcessor<T> create(int capacityHint, Runnable onCancel
      */
     @CheckReturnValue
     @Experimental
+    @NonNull
     public static <T> UnicastProcessor<T> create(int capacityHint, Runnable onCancelled, boolean delayError) {
         ObjectHelper.requireNonNull(onCancelled, "onTerminate");
         return new UnicastProcessor<T>(capacityHint, onCancelled, delayError);

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

@@ -14,6 +14,7 @@
 package io.reactivex.subjects;
 
 import io.reactivex.annotations.Nullable;
+import io.reactivex.annotations.NonNull;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -129,6 +130,7 @@ public final class AsyncSubject<T> extends Subject<T> {
      * @return the new AsyncProcessor instance
      */
     @CheckReturnValue
+    @NonNull
     public static <T> AsyncSubject<T> create() {
         return new AsyncSubject<T>();
     }

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

@@ -15,6 +15,7 @@
 
 import io.reactivex.annotations.CheckReturnValue;
 import io.reactivex.annotations.Nullable;
+import io.reactivex.annotations.NonNull;
 import java.lang.reflect.Array;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.*;
@@ -180,6 +181,7 @@ public final class BehaviorSubject<T> extends Subject<T> {
      * @return the constructed {@link BehaviorSubject}
      */
     @CheckReturnValue
+    @NonNull
     public static <T> BehaviorSubject<T> create() {
         return new BehaviorSubject<T>();
     }
@@ -196,6 +198,7 @@ public static <T> BehaviorSubject<T> create() {
      * @return the constructed {@link BehaviorSubject}
      */
     @CheckReturnValue
+    @NonNull
     public static <T> BehaviorSubject<T> createDefault(T defaultValue) {
         return new BehaviorSubject<T>(defaultValue);
     }

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

@@ -18,6 +18,7 @@
 
 import io.reactivex.*;
 import io.reactivex.annotations.CheckReturnValue;
+import io.reactivex.annotations.NonNull;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -101,6 +102,7 @@ public final class CompletableSubject extends Completable implements Completable
      * @return the new CompletableSubject instance
      */
     @CheckReturnValue
+    @NonNull
     public static CompletableSubject create() {
         return new CompletableSubject();
     }

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

@@ -129,6 +129,7 @@ public final class MaybeSubject<T> extends Maybe<T> implements MaybeObserver<T>
      * @return the new MaybeSubject instance
      */
     @CheckReturnValue
+    @NonNull
     public static <T> MaybeSubject<T> create() {
         return new MaybeSubject<T>();
     }

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

@@ -15,6 +15,7 @@
 
 import io.reactivex.annotations.CheckReturnValue;
 import io.reactivex.annotations.Nullable;
+import io.reactivex.annotations.NonNull;
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.Observer;
@@ -114,6 +115,7 @@ public final class PublishSubject<T> extends Subject<T> {
      * @return the new PublishSubject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> PublishSubject<T> create() {
         return new PublishSubject<T>();
     }

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

@@ -158,6 +158,7 @@ public final class ReplaySubject<T> extends Subject<T> {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplaySubject<T> create() {
         return new ReplaySubject<T>(new UnboundedReplayBuffer<T>(16));
     }
@@ -178,6 +179,7 @@ public static <T> ReplaySubject<T> create() {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplaySubject<T> create(int capacityHint) {
         return new ReplaySubject<T>(new UnboundedReplayBuffer<T>(capacityHint));
     }
@@ -203,6 +205,7 @@ public static <T> ReplaySubject<T> create(int capacityHint) {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplaySubject<T> createWithSize(int maxSize) {
         return new ReplaySubject<T>(new SizeBoundReplayBuffer<T>(maxSize));
     }
@@ -257,6 +260,7 @@ public static <T> ReplaySubject<T> createWithSize(int maxSize) {
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplaySubject<T> createWithTime(long maxAge, TimeUnit unit, Scheduler scheduler) {
         return new ReplaySubject<T>(new SizeAndTimeBoundReplayBuffer<T>(Integer.MAX_VALUE, maxAge, unit, scheduler));
     }
@@ -296,6 +300,7 @@ public static <T> ReplaySubject<T> createWithTime(long maxAge, TimeUnit unit, Sc
      * @return the created subject
      */
     @CheckReturnValue
+    @NonNull
     public static <T> ReplaySubject<T> createWithTimeAndSize(long maxAge, TimeUnit unit, Scheduler scheduler, int maxSize) {
         return new ReplaySubject<T>(new SizeAndTimeBoundReplayBuffer<T>(maxSize, maxAge, unit, scheduler));
     }