2.x: coverage, bugfixes, 9/03-1

Author: akarnokd

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

@@ -1331,15 +1331,15 @@ public final Completable startWith(CompletableSource other) {
     }
 
     /**
-     * Returns an NbpObservable which first delivers the events
-     * of the other NbpObservable then runs this CompletableConsumable.
+     * Returns an Observable which first delivers the events
+     * of the other Observable then runs this CompletableConsumable.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code startWith} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
      * @param <T> the value type
-     * @param other the other NbpObservable to run first
-     * @return the new NbpObservable instance
+     * @param other the other Observable to run first
+     * @return the new Observable instance
      * @throws NullPointerException if other is null
      */
     @SchedulerSupport(SchedulerSupport.NONE)
@@ -1641,14 +1641,14 @@ public final <T> Maybe<T> toMaybe() {
     }
 
     /**
-     * Returns an NbpObservable which when subscribed to subscribes to this Completable and
+     * Returns an Observable which when subscribed to subscribes to this Completable and
      * relays the terminal events to the subscriber.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code toObservable} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
      * @param <T> the value type
-     * @return the new NbpObservable created
+     * @return the new Observable created
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <T> Observable<T> toObservable() {

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

@@ -269,6 +269,9 @@ public static <T, R> Observable<R> combineLatest(ObservableSource<? extends T>[]
     public static <T, R> Observable<R> combineLatest(ObservableSource<? extends T>[] sources, 
             Function<? super T[], ? extends R> combiner, int bufferSize) {
         ObjectHelper.requireNonNull(sources, "sources is null");
+        if (sources.length == 0) {
+            return empty();
+        }
         ObjectHelper.requireNonNull(combiner, "combiner is null");
         ObjectHelper.verifyPositive(bufferSize, "bufferSize");
 
@@ -954,7 +957,7 @@ public static <T> Observable<T> concat(
      * Note: named this way because of overload conflict with concat(ObservableSource&lt;ObservableSource&gt;)
      * @param sources the array of sources
      * @param <T> the common base value type
-     * @return the new NbpObservable instance
+     * @return the new Observable instance
      * @throws NullPointerException if sources is null
      */
     @SuppressWarnings({ "unchecked", "rawtypes" })
@@ -1832,9 +1835,16 @@ public static Observable<Long> intervalRange(long start, long count, long initia
      */
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public static Observable<Long> intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
-
+        if (count < 0) {
+            throw new IllegalArgumentException("count >= 0 required but it was " + count);
+        }
+        
+        if (count == 0L) {
+            return Observable.<Long>empty().delay(initialDelay, unit, scheduler);
+        }
+        
         long end = start + (count - 1);
-        if (end < 0) {
+        if (start > 0 && end < 0) {
             throw new IllegalArgumentException("Overflow! start + count is bigger than Long.MAX_VALUE");
         }
         ObjectHelper.requireNonNull(unit, "unit is null");
@@ -9818,11 +9828,25 @@ public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super T
 
     @Override
     public final void subscribe(Observer<? super T> observer) {
-        ObjectHelper.requireNonNull(observer, "observer is null");
-        
-        observer = RxJavaPlugins.onSubscribe(this, observer);
-        
-        subscribeActual(observer);
+        ObjectHelper.requireNonNull(observer, "s is null");
+        try {
+            observer = RxJavaPlugins.onSubscribe(this, observer);
+
+            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
+            
+            subscribeActual(observer);
+        } catch (NullPointerException e) { // NOPMD
+            throw e;
+        } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
+            // can't call onError because no way to know if a Disposable has been set or not
+            // can't call onSubscribe because the call might have set a Subscription already
+            RxJavaPlugins.onError(e);
+            
+            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
+            npe.initCause(e);
+            throw npe;
+        }
     }
 
     /**
@@ -10151,7 +10175,7 @@ public final Observable<T> takeFirst(Predicate<? super T> predicate) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Observable<T> takeLast(int count) {
         if (count < 0) {
-            throw new IndexOutOfBoundsException("count >= required but it was " + count);
+            throw new IndexOutOfBoundsException("count >= 0 required but it was " + count);
         } else
         if (count == 0) {
             return ignoreElements();

src/main/java/io/reactivex/Observable.java

@@ -31,15 +31,15 @@ private Functions() {
     }
 
     @SuppressWarnings("unchecked")
-    public static <T1, T2, R> Function<Object[], R> toFunction(final BiFunction<? super T1, ? super T2, ? extends R> biFunction) {
-        ObjectHelper.requireNonNull(biFunction, "biFunction is null");
+    public static <T1, T2, R> Function<Object[], R> toFunction(final BiFunction<? super T1, ? super T2, ? extends R> f) {
+        ObjectHelper.requireNonNull(f, "f is null");
         return new Function<Object[], R>() {
             @Override
             public R apply(Object[] a) throws Exception {
                 if (a.length != 2) {
                     throw new IllegalArgumentException("Array of size 2 expected but got " + a.length);
                 }
-                return ((BiFunction<Object, Object, R>)biFunction).apply(a[0], a[1]);
+                return ((BiFunction<Object, Object, R>)f).apply(a[0], a[1]);
             }
         };
     }

src/main/java/io/reactivex/internal/functions/Functions.java

@@ -77,6 +77,8 @@ static final class PublisherBufferExactSubscriber<T, C extends Collection<? supe
         Subscription s;
 
         boolean done;
+        
+        int index;
 
         public PublisherBufferExactSubscriber(Subscriber<? super C> actual, int size, Callable<C> bufferSupplier) {
             this.actual = actual;
@@ -133,10 +135,14 @@ public void onNext(T t) {
             }
 
             b.add(t);
-
-            if (b.size() == size) {
+            
+            int i = index + 1;
+            if (i == size) {
+                index = 0;
                 buffer = null;
                 actual.onNext(b);
+            } else {
+                index = i;
             }
         }
 

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

@@ -40,14 +40,14 @@ public final class FlowableCombineLatest<T, R>
 
     final Iterable<? extends Publisher<? extends T>> iterable;
 
-    final Function<Object[], ? extends R> combiner;
+    final Function<? super T[], ? extends R> combiner;
 
     final int bufferSize;
 
     final boolean delayErrors;
 
     public FlowableCombineLatest(Publisher<? extends T>[] array,
-            Function<Object[], ? extends R> combiner,
+            Function<? super T[], ? extends R> combiner,
                     int bufferSize, boolean delayErrors) {
         if (bufferSize <= 0) {
             throw new IllegalArgumentException("BUFFER_SIZE > 0 required but it was " + bufferSize);
@@ -61,7 +61,7 @@ public FlowableCombineLatest(Publisher<? extends T>[] array,
     }
 
     public FlowableCombineLatest(Iterable<? extends Publisher<? extends T>> iterable,
-            Function<Object[], ? extends R> combiner,
+            Function<? super T[], ? extends R> combiner,
                     int bufferSize, boolean delayErrors) {
         if (bufferSize <= 0) {
             throw new IllegalArgumentException("BUFFER_SIZE > 0 required but it was " + bufferSize);
@@ -150,7 +150,7 @@ public void subscribeActual(Subscriber<? super R> s) {
             new FlowableMap<T, R>((Publisher<T>)a[0], new Function<T, R>() {
                 @Override
                 public R apply(T t) throws Exception {
-                    return combiner.apply(new Object[] { t });
+                    return combiner.apply((T[])new Object[] { t });
                 }
             }).subscribe(s);
             return;
@@ -173,7 +173,7 @@ static final class CombineLatestCoordinator<T, R>
 
         final Subscriber<? super R> actual;
 
-        final Function<Object[], ? extends R> combiner;
+        final Function<? super T[], ? extends R> combiner;
 
         final CombineLatestInnerSubscriber<T>[] subscribers;
 
@@ -198,7 +198,7 @@ static final class CombineLatestCoordinator<T, R>
         final AtomicReference<Throwable> error;
 
         public CombineLatestCoordinator(Subscriber<? super R> actual, 
-                Function<Object[], ? extends R> combiner, int n,
+                Function<? super T[], ? extends R> combiner, int n,
                 int bufferSize, boolean delayErrors) {
             this.actual = actual;
             this.combiner = combiner;

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

@@ -179,6 +179,15 @@ void drain() {
                         e++;
                     }
 
+                    if (e == r) {
+                        boolean d = done;
+                        boolean empty = q.isEmpty();
+                        
+                        if (checkTerminated(d, empty, a)) {
+                            return;
+                        }
+                    }
+                    
                     if (e != 0L) {
                         if (r != Long.MAX_VALUE) {
                             requested.addAndGet(-e);

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

@@ -180,7 +180,7 @@ public static <T> ConnectableFlowable<T> create(Flowable<T> source,
      * @param unit the unit of measure of the age amount
      * @param scheduler the target scheduler providing the current time
      * @param bufferSize the maximum number of elements to hold
-     * @return the new NbpConnectableObservable instance
+     * @return the new ConnectableFlowable instance
      */
     public static <T> ConnectableFlowable<T> create(Flowable<T> source,
             final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize) {