Add @Nullable annotations to Processors

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

@@ -138,6 +138,7 @@ public boolean hasComplete() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         return subscribers.get() == TERMINATED ? error : null;
     }
@@ -244,6 +245,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/processors/BehaviorProcessor.java

@@ -348,6 +348,7 @@ public boolean hasSubscribers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         Object o = value.get();
         if (NotificationLite.isError(o)) {
@@ -361,6 +362,7 @@ public Throwable getThrowable() {
      * <p>The method is thread-safe.
      * @return a single value the BehaviorProcessor 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/processors/PublishProcessor.java

@@ -261,6 +261,7 @@ public boolean hasSubscribers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         if (subscribers.get() == TERMINATED) {
             return error;

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

@@ -354,6 +354,7 @@ public boolean hasSubscribers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         ReplayBuffer<T> b = buffer;
         if (b.isDone()) {
@@ -510,6 +511,7 @@ interface ReplayBuffer<T> {
 
         int size();
 
+        @Nullable
         T getValue();
 
         T[] getValues(T[] array);
@@ -598,6 +600,7 @@ public void trimHead() {
         }
 
         @Override
+        @Nullable
         public T getValue() {
             int s = size;
             if (s == 0) {
@@ -1091,6 +1094,7 @@ public void complete() {
         }
 
         @Override
+        @Nullable
         public T getValue() {
             TimedNode<T> h = head;
 

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

@@ -13,6 +13,7 @@
 
 package io.reactivex.processors;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.internal.util.*;
@@ -187,6 +188,7 @@ public boolean hasThrowable() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         return actual.getThrowable();
     }

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

@@ -457,6 +457,7 @@ public boolean hasSubscribers() {
     }
 
     @Override
+    @Nullable
     public Throwable getThrowable() {
         if (done) {
             return error;