2.x: add withLatestFrom many, cleanups and other enhancements (#4368)

Author: akarnokd

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

@@ -188,6 +188,7 @@ public static Completable create(CompletableSource source) {
         } catch (NullPointerException ex) { // NOPMD
             throw ex;
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             RxJavaPlugins.onError(ex);
             throw toNpe(ex);
         }
@@ -219,6 +220,7 @@ public static Completable unsafeCreate(CompletableSource source) {
         } catch (NullPointerException ex) { // NOPMD
             throw ex;
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             RxJavaPlugins.onError(ex);
             throw toNpe(ex);
         } 
@@ -1384,6 +1386,7 @@ public final void subscribe(CompletableObserver s) {
         } catch (NullPointerException ex) { // NOPMD
             throw ex;
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             RxJavaPlugins.onError(ex);
             throw toNpe(ex);
         }
@@ -1606,6 +1609,7 @@ public final <U> U to(Function<? super Completable, U> converter) {
         try {
             return converter.apply(this);
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             throw Exceptions.propagate(ex);
         }
     }

src/main/java/io/reactivex/CompletableObserver.java

@@ -19,6 +19,13 @@
  * Represents the subscription API callbacks when subscribing to a Completable instance.
  */
 public interface CompletableObserver {
+    /**
+     * Called once by the Completable to set a Disposable on this instance which
+     * then can be used to cancel the subscription at any time.
+     * @param d the Disposable instance to call dispose on for cancellation, not null
+     */
+    void onSubscribe(Disposable d);
+
     /**
      * Called once the deferred computation completes normally.
      */
@@ -29,11 +36,4 @@ public interface CompletableObserver {
      * @param e the exception, not null.
      */
     void onError(Throwable e);
-    
-    /**
-     * Called once by the Completable to set a Disposable on this instance which
-     * then can be used to cancel the subscription at any time.
-     * @param d the Disposable instance to call dispose on for cancellation, not null
-     */
-    void onSubscribe(Disposable d);
 }

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

@@ -15,14 +15,62 @@
 
 import io.reactivex.disposables.Disposable;
 
+/**
+ * Provides a mechanism for receiving push-based notifications.
+ * <p>
+ * After an Observer calls an {@link Observable}'s {@link Observable#subscribe subscribe} method, 
+ * first the Observable calls {@link #onSubscribe(Disposable)} with a {@link Disposable} that allows
+ * cancelling the sequence at any time, then the
+ * {@code Observable} may call the Observer's {@link #onNext} method any number of times
+ * to provide notifications. A well-behaved
+ * {@code Observable} will call an Observer's {@link #onComplete} method exactly once or the Observer's
+ * {@link #onError} method exactly once.
+ * 
+ * @see <a href="http://reactivex.io/documentation/observable.html">ReactiveX documentation: Observable</a>
+ * @param <T>
+ *          the type of item the Observer expects to observe
+ */
 public interface Observer<T> {
-    
+
+    /**
+     * Provides the Observer with the means of cancelling (disposing) the
+     * connection (channel) with the Observable in both
+     * synchronous (from within {@link #onNext(Object)}) and asynchronous manner. 
+     * @param d the Disposable instance whose {@link Disposable#dispose()} can
+     * be called anytime to cancel the connection
+     * @since 2.0
+     */
     void onSubscribe(Disposable d);
 
+    /**
+     * Provides the Observer with a new item to observe.
+     * <p>
+     * The {@link Observable} may call this method 0 or more times.
+     * <p>
+     * The {@code Observable} will not call this method again after it calls either {@link #onComplete} or
+     * {@link #onError}.
+     * 
+     * @param value
+     *          the item emitted by the Observable
+     */
     void onNext(T value);
 
+    /**
+     * Notifies the Observer that the {@link Observable} has experienced an error condition.
+     * <p>
+     * If the {@link Observable} calls this method, it will not thereafter call {@link #onNext} or
+     * {@link #onComplete}.
+     * 
+     * @param e
+     *          the exception encountered by the Observable
+     */
     void onError(Throwable e);
 
+    /**
+     * Notifies the Observer that the {@link Observable} has finished sending push-based notifications.
+     * <p>
+     * The {@link Observable} will not call this method if it calls {@link #onError}.
+     */
     void onComplete();
 
 }

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

@@ -15,8 +15,9 @@
 
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.disposables.*;
+import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
+import io.reactivex.internal.disposables.SequentialDisposable;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public abstract class Scheduler {
@@ -106,9 +107,9 @@ public Disposable schedule(Runnable run) {
         }
 
         public Disposable schedulePeriodically(Runnable run, final long initialDelay, final long period, final TimeUnit unit) {
-            final SerialDisposable first = new SerialDisposable();
+            final SequentialDisposable first = new SequentialDisposable();
 
-            final SerialDisposable sd = new SerialDisposable(first);
+            final SequentialDisposable sd = new SequentialDisposable(first);
 
             final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
 
@@ -182,6 +183,7 @@ public void run() {
                 try {
                     run.run();
                 } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
                     worker.dispose();
                     throw Exceptions.propagate(ex);
                 }

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

@@ -2783,6 +2783,7 @@ public final <R> R to(Function<? super Single<T>, R> convert) {
         try {
             return convert.apply(this);
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             throw Exceptions.propagate(ex);
         }
     }

src/main/java/io/reactivex/Scheduler.java

@@ -15,11 +15,50 @@
 
 import io.reactivex.disposables.Disposable;
 
+/**
+ * Provides a mechanism for receiving push-based notifications.
+ * <p>
+ * After a SingleSubscriber calls a {@link Single}'s {@link Single#subscribe subscribe} method, 
+ * first the Single calls {@link #onSubscribe(Disposable)} with a {@link Disposable} that allows
+ * cancelling the sequence at any time, then the
+ * {@code Single} calls only one of the SingleSubscriber's {@link #onSuccess} and {@link #onError} methods to provide
+ * notifications.
+ * 
+ * @see <a href="http://reactivex.io/documentation/observable.html">ReactiveX documentation: Observable</a>
+ * @param <T>
+ *          the type of item the SingleSubscriber expects to observe
+ * @since 2.0
+ */
 public interface SingleObserver<T> {
 
+    /**
+     * Provides the SingleObserver with the means of cancelling (disposing) the
+     * connection (channel) with the Single in both
+     * synchronous (from within {@link #onSubscribe(Disposable)} itself) and asynchronous manner. 
+     * @param d the Disposable instance whose {@link Disposable#dispose()} can
+     * be called anytime to cancel the connection
+     * @since 2.0
+     */
     void onSubscribe(Disposable d);
 
+    /**
+     * Notifies the SingleSubscriber with a single item and that the {@link Single} has finished sending
+     * push-based notifications.
+     * <p>
+     * The {@link Single} will not call this method if it calls {@link #onError}.
+     * 
+     * @param value
+     *          the item emitted by the Single
+     */
     void onSuccess(T value);
 
+    /**
+     * Notifies the SingleSubscriber that the {@link Single} has experienced an error condition.
+     * <p>
+     * If the {@link Single} calls this method, it will not thereafter call {@link #onSuccess}.
+     * 
+     * @param e
+     *          the exception encountered by the Single
+     */
     void onError(Throwable e);
 }

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

@@ -0,0 +1,20 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Annotations for indicating experimental and beta operators, classes, methods, types or fields.
+ */
+package io.reactivex.annotations;

src/main/java/io/reactivex/SingleObserver.java

@@ -28,6 +28,7 @@ protected void onDisposed(Action value) {
         try {
             value.run();
         } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
             throw Exceptions.propagate(ex);
         }
     }