2.x: More usage of ObjectHelper (#4526)

Author: vanniktech

src/main/java/io/reactivex/Completable.java

@@ -161,9 +161,7 @@ public static Completable concat(Publisher<? extends CompletableSource> sources)
     @SchedulerSupport(SchedulerSupport.NONE)
     public static Completable concat(Publisher<? extends CompletableSource> sources, int prefetch) {
         ObjectHelper.requireNonNull(sources, "sources is null");
-        if (prefetch < 1) {
-            throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);
-        }
+        ObjectHelper.verifyPositive(prefetch, "prefetch");
         return RxJavaPlugins.onAssembly(new CompletableConcat(sources, prefetch));
     }
 
@@ -477,9 +475,7 @@ public static Completable merge(Publisher<? extends CompletableSource> sources,
     @SchedulerSupport(SchedulerSupport.NONE)
     private static Completable merge0(Publisher<? extends CompletableSource> sources, int maxConcurrency, boolean delayErrors) {
         ObjectHelper.requireNonNull(sources, "sources is null");
-        if (maxConcurrency < 1) {
-            throw new IllegalArgumentException("maxConcurrency > 0 required but it was " + maxConcurrency);
-        }
+        ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
         return RxJavaPlugins.onAssembly(new CompletableMerge(sources, maxConcurrency, delayErrors));
     }
 

src/main/java/io/reactivex/Flowable.java

@@ -5399,6 +5399,9 @@ public final Flowable<List<T>> buffer(int count, int skip) {
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U extends Collection<? super T>> Flowable<U> buffer(int count, int skip, Callable<U> bufferSupplier) {
+        ObjectHelper.verifyPositive(count, "count");
+        ObjectHelper.verifyPositive(skip, "skip");
+        ObjectHelper.requireNonNull(bufferSupplier, "bufferSupplier is null");
         return RxJavaPlugins.onAssembly(new FlowableBuffer<T, U>(this, count, skip, bufferSupplier));
     }
 
@@ -6535,6 +6538,7 @@ public final <U> Flowable<U> concatMapIterable(Function<? super T, ? extends Ite
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> Flowable<U> concatMapIterable(final Function<? super T, ? extends Iterable<? extends U>> mapper, int prefetch) {
         ObjectHelper.requireNonNull(mapper, "mapper is null");
+        ObjectHelper.verifyPositive(prefetch, "prefetch");
         return RxJavaPlugins.onAssembly(new FlowableFlattenIterable<T, U>(this, mapper, prefetch));
     }
 
@@ -8198,6 +8202,8 @@ public final <U> Flowable<U> flatMapIterable(final Function<? super T, ? extends
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <U> Flowable<U> flatMapIterable(final Function<? super T, ? extends Iterable<? extends U>> mapper, int bufferSize) {
+        ObjectHelper.requireNonNull(mapper, "mapper is null");
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
         return RxJavaPlugins.onAssembly(new FlowableFlattenIterable<T, U>(this, mapper, bufferSize));
     }
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableAutoConnect.java

@@ -13,10 +13,10 @@
 
 package io.reactivex.internal.operators.flowable;
 
-import java.util.concurrent.atomic.AtomicInteger;
-
 import org.reactivestreams.Subscriber;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.flowables.ConnectableFlowable;
@@ -37,9 +37,6 @@ public final class FlowableAutoConnect<T> extends Flowable<T> {
     public FlowableAutoConnect(ConnectableFlowable<? extends T> source,
             int numberOfSubscribers,
             Consumer<? super Disposable> connection) {
-        if (numberOfSubscribers <= 0) {
-            throw new IllegalArgumentException("numberOfSubscribers > 0 required");
-        }
         this.source = source;
         this.numberOfSubscribers = numberOfSubscribers;
         this.connection = connection;

src/main/java/io/reactivex/internal/operators/flowable/FlowableBuffer.java

@@ -33,23 +33,11 @@ public final class FlowableBuffer<T, C extends Collection<? super T>> extends Ab
 
     final Callable<C> bufferSupplier;
 
-    public FlowableBuffer(Publisher<T> source, int size, Callable<C> bufferSupplier) {
-        this(source, size, size, bufferSupplier);
-    }
-
     public FlowableBuffer(Publisher<T> source, int size, int skip, Callable<C> bufferSupplier) {
         super(source);
-        if (size <= 0) {
-            throw new IllegalArgumentException("size > 0 required but it was " + size);
-        }
-
-        if (skip <= 0) {
-            throw new IllegalArgumentException("skip > 0 required but it was " + size);
-        }
-
         this.size = size;
         this.skip = skip;
-        this.bufferSupplier = ObjectHelper.requireNonNull(bufferSupplier, "bufferSupplier");
+        this.bufferSupplier = bufferSupplier;
     }
 
     @Override

src/main/java/io/reactivex/internal/operators/flowable/FlowableCache.java

@@ -20,6 +20,7 @@
 import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -53,9 +54,7 @@ public static <T> Flowable<T> from(Flowable<T> source) {
      * @return the CachedObservable instance
      */
     public static <T> Flowable<T> from(Flowable<T> source, int capacityHint) {
-        if (capacityHint < 1) {
-            throw new IllegalArgumentException("capacityHint > 0 required");
-        }
+        ObjectHelper.verifyPositive(capacityHint, "capacityHint");
         CacheState<T> state = new CacheState<T>(source, capacityHint);
         return RxJavaPlugins.onAssembly(new FlowableCache<T>(source, state));
     }

src/main/java/io/reactivex/internal/operators/flowable/FlowableCombineLatest.java

@@ -49,27 +49,19 @@ public final class FlowableCombineLatest<T, R>
     public FlowableCombineLatest(Publisher<? extends T>[] array,
             Function<? super Object[], ? extends R> combiner,
                     int bufferSize, boolean delayErrors) {
-        if (bufferSize <= 0) {
-            throw new IllegalArgumentException("BUFFER_SIZE > 0 required but it was " + bufferSize);
-        }
-
-        this.array = ObjectHelper.requireNonNull(array, "array");
+        this.array = array;
         this.iterable = null;
-        this.combiner = ObjectHelper.requireNonNull(combiner, "combiner");
+        this.combiner = combiner;
         this.bufferSize = bufferSize;
         this.delayErrors = delayErrors;
     }
 
     public FlowableCombineLatest(Iterable<? extends Publisher<? extends T>> iterable,
             Function<? super Object[], ? extends R> combiner,
                     int bufferSize, boolean delayErrors) {
-        if (bufferSize <= 0) {
-            throw new IllegalArgumentException("BUFFER_SIZE > 0 required but it was " + bufferSize);
-        }
-
         this.array = null;
-        this.iterable = ObjectHelper.requireNonNull(iterable, "iterable");
-        this.combiner = ObjectHelper.requireNonNull(combiner, "combiner");
+        this.iterable = iterable;
+        this.combiner = combiner;
         this.bufferSize = bufferSize;
         this.delayErrors = delayErrors;
     }

src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java

@@ -38,12 +38,9 @@ public FlowableConcatMap(Publisher<T> source,
             Function<? super T, ? extends Publisher<? extends R>> mapper,
             int prefetch, ErrorMode errorMode) {
         super(source);
-        if (prefetch <= 0) {
-            throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);
-        }
-        this.mapper = ObjectHelper.requireNonNull(mapper, "mapper");
+        this.mapper = mapper;
         this.prefetch = prefetch;
-        this.errorMode = ObjectHelper.requireNonNull(errorMode, "errorMode");
+        this.errorMode = errorMode;
     }
 
     public static <T, R> Subscriber<T> subscribe(Subscriber<? super R> s, Function<? super T, ? extends Publisher<? extends R>> mapper,

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlattenIterable.java

@@ -37,10 +37,7 @@ public final class FlowableFlattenIterable<T, R> extends AbstractFlowableWithUps
     public FlowableFlattenIterable(Publisher<T> source,
             Function<? super T, ? extends Iterable<? extends R>> mapper, int prefetch) {
         super(source);
-        if (prefetch <= 0) {
-            throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);
-        }
-        this.mapper = ObjectHelper.requireNonNull(mapper, "mapper");
+        this.mapper = mapper;
         this.prefetch = prefetch;
     }
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableObserveOn.java

@@ -39,10 +39,7 @@ public FlowableObserveOn(
             boolean delayError,
             int prefetch) {
         super(source);
-        if (prefetch <= 0) {
-            throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);
-        }
-        this.scheduler = ObjectHelper.requireNonNull(scheduler, "scheduler");
+        this.scheduler = scheduler;
         this.delayError = delayError;
         this.prefetch = prefetch;
     }

src/main/java/io/reactivex/internal/operators/observable/ObservableCache.java

@@ -18,6 +18,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
+import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.disposables.SequentialDisposable;
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -52,9 +53,7 @@ public static <T> Observable<T> from(Observable<T> source) {
      * @return the CachedObservable instance
      */
     public static <T> Observable<T> from(Observable<T> source, int capacityHint) {
-        if (capacityHint < 1) {
-            throw new IllegalArgumentException("capacityHint > 0 required");
-        }
+        ObjectHelper.verifyPositive(capacityHint, "capacityHint");
         CacheState<T> state = new CacheState<T>(source, capacityHint);
         return RxJavaPlugins.onAssembly(new ObservableCache<T>(source, state));
     }