add nullable annotation to simple queue (fixes #5053) (#5054)

ReactiveX · Feb 3, 2017 · e5d3b0e · e5d3b0e
1 parent 8720828
commit e5d3b0e
Show file tree

Hide file tree

Showing 55 changed files with 131 additions and 0 deletions.
diff --git a/src/main/java/io/reactivex/internal/disposables/EmptyDisposable.java b/src/main/java/io/reactivex/internal/disposables/EmptyDisposable.java
@@ -14,6 +14,7 @@
 package io.reactivex.internal.disposables;
 
 import io.reactivex.*;
+import io.reactivex.annotations.Nullable;
 import io.reactivex.internal.fuseable.QueueDisposable;
 
 /**
@@ -93,6 +94,7 @@ public boolean offer(Object v1, Object v2) {
         throw new UnsupportedOperationException("Should not be called!");
     }
 
+    @Nullable
     @Override
     public Object poll() throws Exception {
         return null; // always empty

diff --git a/src/main/java/io/reactivex/internal/fuseable/SimplePlainQueue.java b/src/main/java/io/reactivex/internal/fuseable/SimplePlainQueue.java
@@ -13,13 +13,16 @@
 
 package io.reactivex.internal.fuseable;
 
+import io.reactivex.annotations.Nullable;
+
 /**
  * Override of the SimpleQueue interface with no throws Exception on poll.
  *
  * @param <T> the value type to enqueue and dequeue, not null
  */
 public interface SimplePlainQueue<T> extends SimpleQueue<T> {
 
+    @Nullable
     @Override
     T poll();
 }
diff --git a/src/main/java/io/reactivex/internal/fuseable/SimpleQueue.java b/src/main/java/io/reactivex/internal/fuseable/SimpleQueue.java
@@ -13,6 +13,8 @@
 
 package io.reactivex.internal.fuseable;
 
+import io.reactivex.annotations.Nullable;
+
 /**
  * A minimalist queue interface without the method bloat of java.util.Collection and java.util.Queue.
  *
@@ -24,6 +26,10 @@ public interface SimpleQueue<T> {
 
     boolean offer(T v1, T v2);
 
+    /**
+     * @return null to indicate an empty queue
+     */
+    @Nullable
     T poll() throws Exception;
 
     boolean isEmpty();

diff --git a/src/main/java/io/reactivex/internal/observers/DeferredScalarDisposable.java b/src/main/java/io/reactivex/internal/observers/DeferredScalarDisposable.java
@@ -14,6 +14,7 @@
 package io.reactivex.internal.observers;
 
 import io.reactivex.Observer;
+import io.reactivex.annotations.Nullable;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -110,6 +111,7 @@ public final void complete() {
         actual.onComplete();
     }
 
+    @Nullable
     @Override
     public final T poll() throws Exception {
         if (get() == FUSED_READY) {

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableCombineLatest.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableCombineLatest.java
@@ -16,6 +16,7 @@
 import java.util.Iterator;
 import java.util.concurrent.atomic.*;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.Flowable;
@@ -466,6 +467,7 @@ public int requestFusion(int requestedMode) {
             return m;
         }
 
+        @Nullable
         @SuppressWarnings("unchecked")
         @Override
         public R poll() throws Exception {

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDistinct.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDistinct.java
@@ -16,6 +16,7 @@
 import java.util.Collection;
 import java.util.concurrent.Callable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.Exceptions;
@@ -117,6 +118,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             for (;;) {

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDistinctUntilChanged.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDistinctUntilChanged.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.functions.*;
@@ -106,6 +107,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             for (;;) {
@@ -195,6 +197,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             for (;;) {

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDoAfterNext.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDoAfterNext.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.annotations.Experimental;
@@ -74,6 +75,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = qs.poll();
@@ -122,6 +124,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = qs.poll();

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDoFinally.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDoFinally.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.annotations.Experimental;
@@ -130,6 +131,7 @@ public boolean isEmpty() {
             return qs.isEmpty();
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = qs.poll();
@@ -239,6 +241,7 @@ public boolean isEmpty() {
             return qs.isEmpty();
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = qs.poll();

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDoOnEach.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDoOnEach.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.*;
@@ -144,6 +145,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = qs.poll();
@@ -276,6 +278,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = qs.poll();

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFilter.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFilter.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.functions.Predicate;
@@ -79,6 +80,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             QueueSubscription<T> qs = this.qs;
@@ -143,6 +145,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             QueueSubscription<T> qs = this.qs;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapCompletable.java
@@ -15,6 +15,7 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.*;
@@ -174,6 +175,7 @@ public void request(long n) {
             // ignored, no values emitted
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             return null; // always empty

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlattenIterable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlattenIterable.java
@@ -17,6 +17,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.*;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.*;
@@ -413,6 +414,7 @@ public boolean isEmpty() {
             return (it != null && !it.hasNext()) || queue.isEmpty();
         }
 
+        @Nullable
         @Override
         public R poll() throws Exception {
             Iterator<? extends R> it = current;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromArray.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromArray.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.Subscriber;
 
 import io.reactivex.Flowable;
@@ -55,6 +56,7 @@ public final int requestFusion(int mode) {
             return mode & SYNC;
         }
 
+        @Nullable
         @Override
         public final T poll() {
             int i = index;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromIterable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromIterable.java
@@ -15,6 +15,7 @@
 
 import java.util.Iterator;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.Subscriber;
 
 import io.reactivex.Flowable;
@@ -87,6 +88,7 @@ public final int requestFusion(int mode) {
             return mode & SYNC;
         }
 
+        @Nullable
         @Override
         public final T poll() {
             if (it == null) {

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupBy.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupBy.java
@@ -17,6 +17,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.*;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.Exceptions;
@@ -353,6 +354,7 @@ public int requestFusion(int mode) {
             return NONE;
         }
 
+        @Nullable
         @Override
         public GroupedFlowable<K, V> poll() {
             return queue.poll();
@@ -627,6 +629,7 @@ public int requestFusion(int mode) {
             return NONE;
         }
 
+        @Nullable
         @Override
         public T poll() {
             T v = queue.poll();

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableIgnoreElements.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableIgnoreElements.java
@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.internal.fuseable.QueueSubscription;
@@ -72,6 +73,7 @@ public boolean offer(T v1, T v2) {
             throw new UnsupportedOperationException("Should not be called!");
         }
 
+        @Nullable
         @Override
         public T poll() {
             return null; // empty, always

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableMap.java
@@ -14,6 +14,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.functions.Function;
@@ -72,6 +73,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public U poll() throws Exception {
             T t = qs.poll();
@@ -131,6 +133,7 @@ public int requestFusion(int mode) {
             return transitiveBoundaryFusion(mode);
         }
 
+        @Nullable
         @Override
         public U poll() throws Exception {
             T t = qs.poll();

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableObserveOn.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableObserveOn.java
@@ -15,6 +15,7 @@
 
 import java.util.concurrent.atomic.AtomicLong;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.Scheduler;
@@ -457,6 +458,7 @@ void runBackfused() {
             }
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = queue.poll();
@@ -695,6 +697,7 @@ void runBackfused() {
             }
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             T v = queue.poll();

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureBuffer.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureBuffer.java
@@ -15,6 +15,7 @@
 
 import java.util.concurrent.atomic.AtomicLong;
 
+import io.reactivex.annotations.Nullable;
 import org.reactivestreams.*;
 
 import io.reactivex.exceptions.*;
@@ -251,6 +252,7 @@ public int requestFusion(int mode) {
             return NONE;
         }
 
+        @Nullable
         @Override
         public T poll() throws Exception {
             return queue.poll();