Added nullable annotations to subjects (#5890)

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

@@ -13,6 +13,7 @@
 
 package io.reactivex.subjects;
 
+import io.reactivex.annotations.Nullable;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -85,22 +86,22 @@
  * AsyncSubject<Object> subject = AsyncSubject.create();
  * 
  * TestObserver<Object> to1 = subject.test();
- * 
+ *
  * to1.assertEmpty();
- * 
+ *
  * subject.onNext(1);
- * 
+ *
  * // AsyncSubject only emits when onComplete was called.
  * to1.assertEmpty();
  *
  * subject.onNext(2);
  * subject.onComplete();
- * 
+ *
  * // onComplete triggers the emission of the last cached item and the onComplete event.
  * to1.assertResult(2);
- * 
+ *
  * TestObserver<Object> to2 = subject.test();
- * 
+ *
  * // late Observers receive the last cached item too
  * to2.assertResult(2);
  * </code></pre>
@@ -313,6 +314,7 @@ public boolean hasValue() {
      * <p>The method is thread-safe.
      * @return a single value the Subject currently has or null if no such value exists
      */
+    @Nullable
     public T getValue() {
         return subscribers.get() == TERMINATED ? value : null;
     }

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

@@ -14,6 +14,7 @@
 package io.reactivex.subjects;
 
 import io.reactivex.annotations.CheckReturnValue;
+import io.reactivex.annotations.Nullable;
 import java.lang.reflect.Array;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.*;
@@ -63,19 +64,19 @@
  * observable.onNext(1);
  * // this will "clear" the cache
  * observable.onNext(EMPTY);
- * 
+ *
  * TestObserver<Integer> to2 = observable.test();
- * 
+ *
  * subject.onNext(2);
  * subject.onComplete();
- * 
+ *
  * // to1 received both non-empty items
  * to1.assertResult(1, 2);
- * 
+ *
  * // to2 received only 2 even though the current item was EMPTY
  * // when it got subscribed
  * to2.assertResult(2);
- * 
+ *
  * // Observers coming after the subject was terminated receive
  * // no items and only the onComplete event in this case.
  * observable.test().assertResult();
@@ -300,6 +301,7 @@ public boolean hasObservers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         Object o = value.get();
         if (NotificationLite.isError(o)) {
@@ -313,6 +315,7 @@ public Throwable getThrowable() {
      * <p>The method is thread-safe.
      * @return a single value the Subject currently has or null if no such value exists
      */
+    @Nullable
     public T getValue() {
         Object o = value.get();
         if (NotificationLite.isComplete(o) || NotificationLite.isError(o)) {

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

@@ -13,6 +13,7 @@
 
 package io.reactivex.subjects;
 
+import io.reactivex.annotations.Nullable;
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.*;
@@ -65,12 +66,12 @@
  * Example usage:
  * <pre><code>
  * CompletableSubject subject = CompletableSubject.create();
- * 
+ *
  * TestObserver&lt;Void&gt; to1 = subject.test();
  *
  * // a fresh CompletableSubject is empty
  * to1.assertEmpty();
- * 
+ *
  * subject.onComplete();
  *
  * // a CompletableSubject is always void of items
@@ -213,6 +214,7 @@ void remove(CompletableDisposable inner) {
      * Returns the terminal error if this CompletableSubject has been terminated with an error, null otherwise.
      * @return the terminal error or null if not terminated or not with an error
      */
+    @Nullable
     public Throwable getThrowable() {
         if (observers.get() == TERMINATED) {
             return error;

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

@@ -72,20 +72,20 @@
  * Example usage:
  * <pre><code>
  * MaybeSubject&lt;Integer&gt; subject1 = MaybeSubject.create();
- * 
+ *
  * TestObserver&lt;Integer&gt; to1 = subject1.test();
- * 
+ *
  * // MaybeSubjects are empty by default
  * to1.assertEmpty();
- * 
+ *
  * subject1.onSuccess(1);
- * 
+ *
  * // onSuccess is a terminal event with MaybeSubjects
  * // TestObserver converts onSuccess into onNext + onComplete
  * to1.assertResult(1);
  *
  * TestObserver&lt;Integer&gt; to2 = subject1.test();
- * 
+ *
  * // late Observers receive the terminal signal (onSuccess) too
  * to2.assertResult(1);
  *
@@ -94,14 +94,14 @@
  * MaybeSubject&lt;Integer&gt; subject2 = MaybeSubject.create();
  *
  * TestObserver&lt;Integer&gt; to3 = subject2.test();
- * 
+ *
  * subject2.onComplete();
- * 
+ *
  * // a completed MaybeSubject completes its MaybeObservers
  * to3.assertResult();
  *
  * TestObserver&lt;Integer&gt; to4 = subject1.test();
- * 
+ *
  * // late Observers receive the terminal signal (onComplete) too
  * to4.assertResult();
  * </code></pre>
@@ -263,6 +263,7 @@ void remove(MaybeDisposable<T> inner) {
      * Returns the success value if this MaybeSubject was terminated with a success value.
      * @return the success value or null
      */
+    @Nullable
     public T getValue() {
         if (observers.get() == TERMINATED) {
             return value;
@@ -282,6 +283,7 @@ public boolean hasValue() {
      * Returns the terminal error if this MaybeSubject has been terminated with an error, null otherwise.
      * @return the terminal error or null if not terminated or not with an error
      */
+    @Nullable
     public Throwable getThrowable() {
         if (observers.get() == TERMINATED) {
             return error;

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

@@ -14,6 +14,7 @@
 package io.reactivex.subjects;
 
 import io.reactivex.annotations.CheckReturnValue;
+import io.reactivex.annotations.Nullable;
 import java.util.concurrent.atomic.*;
 
 import io.reactivex.Observer;
@@ -263,6 +264,7 @@ public boolean hasObservers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         if (subscribers.get() == TERMINATED) {
             return error;

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

@@ -13,6 +13,7 @@
 
 package io.reactivex.subjects;
 
+import io.reactivex.annotations.Nullable;
 import java.lang.reflect.Array;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
@@ -395,6 +396,7 @@ public boolean hasObservers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         Object o = buffer.get();
         if (NotificationLite.isError(o)) {
@@ -408,6 +410,7 @@ public Throwable getThrowable() {
      * <p>The method is thread-safe.
      * @return a single value the Subject currently has or null if no such value exists
      */
+    @Nullable
     public T getValue() {
         return buffer.getValue();
     }
@@ -542,6 +545,7 @@ interface ReplayBuffer<T> {
 
         int size();
 
+        @Nullable
         T getValue();
 
         T[] getValues(T[] array);
@@ -620,6 +624,7 @@ public void addFinal(Object notificationLite) {
         }
 
         @Override
+        @Nullable
         @SuppressWarnings("unchecked")
         public T getValue() {
             int s = size;
@@ -838,6 +843,7 @@ public void addFinal(Object notificationLite) {
         }
 
         @Override
+        @Nullable
         @SuppressWarnings("unchecked")
         public T getValue() {
             Node<Object> prev = null;
@@ -1080,6 +1086,7 @@ public void addFinal(Object notificationLite) {
         }
 
         @Override
+        @Nullable
         @SuppressWarnings("unchecked")
         public T getValue() {
             TimedNode<Object> prev = null;

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

@@ -14,6 +14,7 @@
 package io.reactivex.subjects;
 
 import io.reactivex.Observer;
+import io.reactivex.annotations.Nullable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.util.*;
 import io.reactivex.internal.util.AppendOnlyLinkedArrayList.NonThrowingPredicate;
@@ -193,6 +194,7 @@ public boolean hasThrowable() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         return actual.getThrowable();
     }

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

@@ -105,37 +105,37 @@
  * Example usage:
  * <pre><code>
  * UnicastSubject&lt;Integer&gt; subject = UnicastSubject.create();
- * 
+ *
  * TestObserver&lt;Integer&gt; to1 = subject.test();
- * 
+ *
  * // fresh UnicastSubjects are empty
  * to1.assertEmpty();
- * 
+ *
  * TestObserver&lt;Integer&gt; to2 = subject.test();
- * 
+ *
  * // A UnicastSubject only allows one Observer during its lifetime
  * to2.assertFailure(IllegalStateException.class);
- * 
+ *
  * subject.onNext(1);
  * to1.assertValue(1);
- * 
+ *
  * subject.onNext(2);
  * to1.assertValues(1, 2);
- * 
+ *
  * subject.onComplete();
  * to1.assertResult(1, 2);
- * 
+ *
  * // ----------------------------------------------------
- * 
+ *
  * UnicastSubject&lt;Integer&gt; subject2 = UnicastSubject.create();
- * 
+ *
  * // a UnicastSubject caches events util its single Observer subscribes
  * subject.onNext(1);
  * subject.onNext(2);
  * subject.onComplete();
- * 
+ *
  * TestObserver&lt;Integer&gt; to3 = subject2.test();
- * 
+ *
  * // the cached events are emitted in order
  * to3.assertResult(1, 2);
  * </code></pre>
@@ -498,6 +498,7 @@ public boolean hasObservers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         if (done) {
             return error;