More nullability annotations

src/main/java/io/reactivex/Observer.java

@@ -13,6 +13,7 @@
 
 package io.reactivex;
 
+import io.reactivex.annotations.NonNull;
 import io.reactivex.disposables.Disposable;
 
 /**
@@ -40,7 +41,7 @@ public interface Observer<T> {
      * be called anytime to cancel the connection
      * @since 2.0
      */
-    void onSubscribe(Disposable d);
+    void onSubscribe(@NonNull Disposable d);
 
     /**
      * Provides the Observer with a new item to observe.
@@ -53,7 +54,7 @@ public interface Observer<T> {
      * @param t
      *          the item emitted by the Observable
      */
-    void onNext(T t);
+    void onNext(@NonNull T t);
 
     /**
      * Notifies the Observer that the {@link Observable} has experienced an error condition.
@@ -64,7 +65,7 @@ public interface Observer<T> {
      * @param e
      *          the exception encountered by the Observable
      */
-    void onError(Throwable e);
+    void onError(@NonNull Throwable e);
 
     /**
      * Notifies the Observer that the {@link Observable} has finished sending push-based notifications.

src/main/java/io/reactivex/Single.java

@@ -13,27 +13,42 @@
 
 package io.reactivex;
 
-import java.util.NoSuchElementException;
-import java.util.concurrent.*;
-
-import org.reactivestreams.Publisher;
-
 import io.reactivex.annotations.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
-import io.reactivex.internal.functions.*;
-import io.reactivex.internal.fuseable.*;
-import io.reactivex.internal.observers.*;
-import io.reactivex.internal.operators.completable.*;
-import io.reactivex.internal.operators.flowable.*;
-import io.reactivex.internal.operators.maybe.*;
-import io.reactivex.internal.operators.observable.*;
+import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.internal.fuseable.FuseToFlowable;
+import io.reactivex.internal.fuseable.FuseToMaybe;
+import io.reactivex.internal.fuseable.FuseToObservable;
+import io.reactivex.internal.observers.BiConsumerSingleObserver;
+import io.reactivex.internal.observers.BlockingMultiObserver;
+import io.reactivex.internal.observers.ConsumerSingleObserver;
+import io.reactivex.internal.observers.FutureSingleObserver;
+import io.reactivex.internal.operators.completable.CompletableFromSingle;
+import io.reactivex.internal.operators.completable.CompletableToFlowable;
+import io.reactivex.internal.operators.flowable.FlowableConcatMap;
+import io.reactivex.internal.operators.flowable.FlowableConcatMapPublisher;
+import io.reactivex.internal.operators.flowable.FlowableFlatMapPublisher;
+import io.reactivex.internal.operators.flowable.FlowableSingleSingle;
+import io.reactivex.internal.operators.maybe.MaybeFilterSingle;
+import io.reactivex.internal.operators.maybe.MaybeFromSingle;
+import io.reactivex.internal.operators.observable.ObservableConcatMap;
+import io.reactivex.internal.operators.observable.ObservableSingleSingle;
 import io.reactivex.internal.operators.single.*;
-import io.reactivex.internal.util.*;
+import io.reactivex.internal.util.ErrorMode;
+import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
+import org.reactivestreams.Publisher;
+
+import java.util.NoSuchElementException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The Single class implements the Reactive Pattern for a single value response.
@@ -2714,7 +2729,7 @@ public final void subscribe(SingleObserver<? super T> subscriber) {
      * Override this method in subclasses to handle the incoming SingleObservers.
      * @param observer the SingleObserver to handle, not null
      */
-    protected abstract void subscribeActual(SingleObserver<? super T> observer);
+    protected abstract void subscribeActual(@NonNull SingleObserver<? super T> observer);
 
     /**
      * Subscribes a given SingleObserver (subclass) to this Single and returns the given

src/main/java/io/reactivex/disposables/ActionDisposable.java

@@ -12,19 +12,20 @@
  */
 package io.reactivex.disposables;
 
+import io.reactivex.annotations.NonNull;
 import io.reactivex.functions.Action;
 import io.reactivex.internal.util.ExceptionHelper;
 
 final class ActionDisposable extends ReferenceDisposable<Action> {
 
     private static final long serialVersionUID = -8219729196779211169L;
 
-    ActionDisposable(Action value) {
+    ActionDisposable(@NonNull Action value) {
         super(value);
     }
 
     @Override
-    protected void onDisposed(Action value) {
+    protected void onDisposed(@NonNull Action value) {
         try {
             value.run();
         } catch (Throwable ex) {

src/main/java/io/reactivex/disposables/CompositeDisposable.java

@@ -12,12 +12,16 @@
  */
 package io.reactivex.disposables;
 
-import java.util.*;
-
-import io.reactivex.exceptions.*;
+import io.reactivex.annotations.NonNull;
+import io.reactivex.exceptions.CompositeException;
+import io.reactivex.exceptions.Exceptions;
 import io.reactivex.internal.disposables.DisposableContainer;
 import io.reactivex.internal.functions.ObjectHelper;
-import io.reactivex.internal.util.*;
+import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.OpenHashSet;
+
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * A disposable container that can hold onto multiple other disposables and
@@ -39,7 +43,7 @@ public CompositeDisposable() {
      * Creates a CompositeDisposables with the given array of initial elements.
      * @param resources the array of Disposables to start with
      */
-    public CompositeDisposable(Disposable... resources) {
+    public CompositeDisposable(@NonNull Disposable... resources) {
         ObjectHelper.requireNonNull(resources, "resources is null");
         this.resources = new OpenHashSet<Disposable>(resources.length + 1);
         for (Disposable d : resources) {
@@ -52,7 +56,7 @@ public CompositeDisposable(Disposable... resources) {
      * Creates a CompositeDisposables with the given Iterable sequence of initial elements.
      * @param resources the Iterable sequence of Disposables to start with
      */
-    public CompositeDisposable(Iterable<? extends Disposable> resources) {
+    public CompositeDisposable(@NonNull Iterable<? extends Disposable> resources) {
         ObjectHelper.requireNonNull(resources, "resources is null");
         this.resources = new OpenHashSet<Disposable>();
         for (Disposable d : resources) {
@@ -85,7 +89,7 @@ public boolean isDisposed() {
     }
 
     @Override
-    public boolean add(Disposable d) {
+    public boolean add(@NonNull Disposable d) {
         ObjectHelper.requireNonNull(d, "d is null");
         if (!disposed) {
             synchronized (this) {
@@ -110,7 +114,7 @@ public boolean add(Disposable d) {
      * @param ds the array of Disposables
      * @return true if the operation was successful, false if the container has been disposed
      */
-    public boolean addAll(Disposable... ds) {
+    public boolean addAll(@NonNull Disposable... ds) {
         ObjectHelper.requireNonNull(ds, "ds is null");
         if (!disposed) {
             synchronized (this) {
@@ -135,7 +139,7 @@ public boolean addAll(Disposable... ds) {
     }
 
     @Override
-    public boolean remove(Disposable d) {
+    public boolean remove(@NonNull Disposable d) {
         if (delete(d)) {
             d.dispose();
             return true;
@@ -144,7 +148,7 @@ public boolean remove(Disposable d) {
     }
 
     @Override
-    public boolean delete(Disposable d) {
+    public boolean delete(@NonNull Disposable d) {
         ObjectHelper.requireNonNull(d, "Disposable item is null");
         if (disposed) {
             return false;

src/main/java/io/reactivex/disposables/Disposables.java

@@ -13,13 +13,14 @@
 
 package io.reactivex.disposables;
 
-import java.util.concurrent.Future;
-
-import org.reactivestreams.Subscription;
-
+import io.reactivex.annotations.NonNull;
 import io.reactivex.functions.Action;
 import io.reactivex.internal.disposables.EmptyDisposable;
-import io.reactivex.internal.functions.*;
+import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.functions.ObjectHelper;
+import org.reactivestreams.Subscription;
+
+import java.util.concurrent.Future;
 
 /**
  * Utility class to help create disposables by wrapping
@@ -38,7 +39,8 @@ private Disposables() {
      * @param run the Runnable to wrap
      * @return the new Disposable instance
      */
-    public static Disposable fromRunnable(Runnable run) {
+    @NonNull
+    public static Disposable fromRunnable(@NonNull Runnable run) {
         ObjectHelper.requireNonNull(run, "run is null");
         return new RunnableDisposable(run);
     }
@@ -49,7 +51,8 @@ public static Disposable fromRunnable(Runnable run) {
      * @param run the Action to wrap
      * @return the new Disposable instance
      */
-    public static Disposable fromAction(Action run) {
+    @NonNull
+    public static Disposable fromAction(@NonNull Action run) {
         ObjectHelper.requireNonNull(run, "run is null");
         return new ActionDisposable(run);
     }
@@ -60,7 +63,8 @@ public static Disposable fromAction(Action run) {
      * @param future the Future to wrap
      * @return the new Disposable instance
      */
-    public static Disposable fromFuture(Future<?> future) {
+    @NonNull
+    public static Disposable fromFuture(@NonNull Future<?> future) {
         ObjectHelper.requireNonNull(future, "future is null");
         return fromFuture(future, true);
     }
@@ -72,7 +76,8 @@ public static Disposable fromFuture(Future<?> future) {
      * @param allowInterrupt if true, the future cancel happens via Future.cancel(true)
      * @return the new Disposable instance
      */
-    public static Disposable fromFuture(Future<?> future, boolean allowInterrupt) {
+    @NonNull
+    public static Disposable fromFuture(@NonNull Future<?> future, boolean allowInterrupt) {
         ObjectHelper.requireNonNull(future, "future is null");
         return new FutureDisposable(future, allowInterrupt);
     }
@@ -83,7 +88,8 @@ public static Disposable fromFuture(Future<?> future, boolean allowInterrupt) {
      * @param subscription the Runnable to wrap
      * @return the new Disposable instance
      */
-    public static Disposable fromSubscription(Subscription subscription) {
+    @NonNull
+    public static Disposable fromSubscription(@NonNull Subscription subscription) {
         ObjectHelper.requireNonNull(subscription, "subscription is null");
         return new SubscriptionDisposable(subscription);
     }
@@ -92,6 +98,7 @@ public static Disposable fromSubscription(Subscription subscription) {
      * Returns a new, non-disposed Disposable instance.
      * @return a new, non-disposed Disposable instance
      */
+    @NonNull
     public static Disposable empty() {
         return fromRunnable(Functions.EMPTY_RUNNABLE);
     }
@@ -100,6 +107,7 @@ public static Disposable empty() {
      * Returns a disposed Disposable instance.
      * @return a disposed Disposable instance
      */
+    @NonNull
     public static Disposable disposed() {
         return EmptyDisposable.INSTANCE;
     }

src/main/java/io/reactivex/disposables/FutureDisposable.java

@@ -12,6 +12,8 @@
  */
 package io.reactivex.disposables;
 
+import io.reactivex.annotations.Nullable;
+
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -24,7 +26,7 @@ final class FutureDisposable extends AtomicReference<Future<?>> implements Dispo
 
     private final boolean allowInterrupt;
 
-    FutureDisposable(Future<?> run, boolean allowInterrupt) {
+    FutureDisposable(@Nullable Future<?> run, boolean allowInterrupt) {
         super(run);
         this.allowInterrupt = allowInterrupt;
     }

src/main/java/io/reactivex/disposables/ReferenceDisposable.java

@@ -13,10 +13,11 @@
 
 package io.reactivex.disposables;
 
-import java.util.concurrent.atomic.AtomicReference;
-
+import io.reactivex.annotations.NonNull;
 import io.reactivex.internal.functions.ObjectHelper;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 /**
  * Base class for Disposable containers that manage some other type that
  * has to be run when the container is disposed.
@@ -27,11 +28,11 @@ abstract class ReferenceDisposable<T> extends AtomicReference<T> implements Disp
 
     private static final long serialVersionUID = 6537757548749041217L;
 
-    ReferenceDisposable(T value) {
+    ReferenceDisposable(@NonNull T value) {
         super(ObjectHelper.requireNonNull(value, "value is null"));
     }
 
-    protected abstract void onDisposed(T value);
+    protected abstract void onDisposed(@NonNull T value);
 
     @Override
     public final void dispose() {

src/main/java/io/reactivex/disposables/RunnableDisposable.java

@@ -12,19 +12,21 @@
  */
 package io.reactivex.disposables;
 
+import io.reactivex.annotations.NonNull;
+
 /**
  * A disposable container that manages a Runnable instance.
  */
 final class RunnableDisposable extends ReferenceDisposable<Runnable> {
 
     private static final long serialVersionUID = -8219729196779211169L;
 
-    RunnableDisposable(Runnable value) {
+    RunnableDisposable(@NonNull Runnable value) {
         super(value);
     }
 
     @Override
-    protected void onDisposed(Runnable value) {
+    protected void onDisposed(@NonNull Runnable value) {
         value.run();
     }