1.x: add Completable.safeSubscribe option + RxJavaPlugins hook support (#3942)

Add option to safely subscribe a CompletableSubscriber / regular Subscriber and handle onXXX failures.

Author: akarnokd

src/main/java/rx/Completable.java

@@ -2024,7 +2024,7 @@ public void onSubscribe(Subscription d) {
                 serial.add(main);
                 child.add(serial);
 
-                other.subscribe(so);
+                other.unsafeSubscribe(so);
 
                 return main;
             }

src/main/java/rx/Single.java

@@ -130,7 +130,7 @@ void next() {
                 return;
             }
 
-            c.subscribe(inner);
+            c.unsafeSubscribe(inner);
         }
 
         final class ConcatInnerSubscriber implements CompletableSubscriber {

src/main/java/rx/internal/operators/CompletableOnSubscribeConcat.java

@@ -89,7 +89,7 @@ void next() {
                     return;
                 }
 
-                a[idx].subscribe(this);
+                a[idx].unsafeSubscribe(this);
             } while (decrementAndGet() != 0);
         }
     }

src/main/java/rx/internal/operators/CompletableOnSubscribeConcatArray.java

@@ -128,7 +128,7 @@ void next() {
                     return;
                 }
 
-                c.subscribe(this);
+                c.unsafeSubscribe(this);
             } while (decrementAndGet() != 0);
         }
     }

src/main/java/rx/internal/operators/CompletableOnSubscribeConcatIterable.java

@@ -101,7 +101,7 @@ public void onNext(Completable t) {
 
             wip.getAndIncrement();
 
-            t.subscribe(new CompletableSubscriber() {
+            t.unsafeSubscribe(new CompletableSubscriber() {
                 Subscription d;
                 boolean innerDone;
                 @Override

src/main/java/rx/internal/operators/CompletableOnSubscribeMerge.java

@@ -54,7 +54,7 @@ public void call(final CompletableSubscriber s) {
                 }
             }
 
-            c.subscribe(new CompletableSubscriber() {
+            c.unsafeSubscribe(new CompletableSubscriber() {
                 @Override
                 public void onSubscribe(Subscription d) {
                     set.add(d);

src/main/java/rx/internal/operators/CompletableOnSubscribeMergeArray.java

@@ -51,7 +51,7 @@ public void call(final CompletableSubscriber s) {
                 continue;
             }
 
-            c.subscribe(new CompletableSubscriber() {
+            c.unsafeSubscribe(new CompletableSubscriber() {
                 @Override
                 public void onSubscribe(Subscription d) {
                     set.add(d);

src/main/java/rx/internal/operators/CompletableOnSubscribeMergeDelayErrorArray.java

@@ -117,7 +117,7 @@ public void call(final CompletableSubscriber s) {
 
             wip.getAndIncrement();
 
-            c.subscribe(new CompletableSubscriber() {
+            c.unsafeSubscribe(new CompletableSubscriber() {
                 @Override
                 public void onSubscribe(Subscription d) {
                     set.add(d);

src/main/java/rx/internal/operators/CompletableOnSubscribeMergeDelayErrorIterable.java

@@ -110,7 +110,7 @@ public void call(final CompletableSubscriber s) {
 
             wip.getAndIncrement();
 
-            c.subscribe(new CompletableSubscriber() {
+            c.unsafeSubscribe(new CompletableSubscriber() {
                 @Override
                 public void onSubscribe(Subscription d) {
                     set.add(d);

src/main/java/rx/internal/operators/CompletableOnSubscribeMergeIterable.java

@@ -60,7 +60,7 @@ public void call() {
                     if (other == null) {
                         s.onError(new TimeoutException());
                     } else {
-                        other.subscribe(new CompletableSubscriber() {
+                        other.unsafeSubscribe(new CompletableSubscriber() {
 
                             @Override
                             public void onSubscribe(Subscription d) {
@@ -85,7 +85,7 @@ public void onCompleted() {
             }
         }, timeout, unit);
 
-        source.subscribe(new CompletableSubscriber() {
+        source.unsafeSubscribe(new CompletableSubscriber() {
 
             @Override
             public void onSubscribe(Subscription d) {

src/main/java/rx/internal/operators/CompletableOnSubscribeTimeout.java

@@ -0,0 +1,94 @@
+/**
+ * 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.
+ */
+package rx.observers;
+
+import rx.Completable.CompletableSubscriber;
+import rx.Subscription;
+import rx.annotations.Experimental;
+import rx.exceptions.*;
+import rx.internal.util.RxJavaPluginUtils;
+
+/**
+ * Wraps another CompletableSubscriber and handles exceptions thrown
+ * from onError and onCompleted.
+ * 
+ * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+ */
+@Experimental
+public final class SafeCompletableSubscriber implements CompletableSubscriber, Subscription {
+    final CompletableSubscriber actual;
+
+    Subscription s;
+    
+    boolean done;
+    
+    public SafeCompletableSubscriber(CompletableSubscriber actual) {
+        this.actual = actual;
+    }
+
+    @Override
+    public void onCompleted() {
+        if (done) {
+            return;
+        }
+        done = true;
+        try {
+            actual.onCompleted();
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            
+            throw new OnCompletedFailedException(ex);
+        }
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        RxJavaPluginUtils.handleException(e);
+        if (done) {
+            return;
+        }
+        done = true;
+        try {
+            actual.onError(e);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            
+            throw new OnErrorFailedException(new CompositeException(e, ex));
+        }
+    }
+
+    @Override
+    public void onSubscribe(Subscription d) {
+        this.s = d;
+        try {
+            actual.onSubscribe(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            d.unsubscribe();
+            onError(ex);
+        }
+    }
+    
+    @Override
+    public void unsubscribe() {
+        s.unsubscribe();
+    }
+    
+    @Override
+    public boolean isUnsubscribed() {
+        return done || s.isUnsubscribed();
+    }
+}

src/main/java/rx/observers/SafeCompletableSubscriber.java

@@ -0,0 +1,104 @@
+/**
+ * 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.
+ */
+package rx.plugins;
+
+import rx.*;
+import rx.annotations.Experimental;
+import rx.functions.Func1;
+
+/**
+ * Abstract ExecutionHook with invocations at different lifecycle points of {@link Completable} execution with a
+ * default no-op implementation.
+ * <p>
+ * See {@link RxJavaPlugins} or the RxJava GitHub Uncyclo for information on configuring plugins:
+ * <a href="https://github.com/ReactiveX/RxJava/wiki/Plugins">https://github.com/ReactiveX/RxJava/wiki/Plugins</a>.
+ * <p>
+ * <b>Note on thread-safety and performance:</b>
+ * <p>
+ * A single implementation of this class will be used globally so methods on this class will be invoked
+ * concurrently from multiple threads so all functionality must be thread-safe.
+ * <p>
+ * Methods are also invoked synchronously and will add to execution time of the completable so all behavior
+ * should be fast. If anything time-consuming is to be done it should be spawned asynchronously onto separate
+ * worker threads.
+ *
+ * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+ */
+@Experimental
+public abstract class RxJavaCompletableExecutionHook {
+    /**
+     * Invoked during the construction by {@link Completable#create(Completable.CompletableOnSubscribe)}
+     * <p>
+     * This can be used to decorate or replace the <code>onSubscribe</code> function or just perform extra
+     * logging, metrics and other such things and pass through the function.
+     *
+     * @param f
+     *            original {@link rx.Completable.CompletableOnSubscribe}<{@code T}> to be executed
+     * @return {@link rx.Completable.CompletableOnSubscribe} function that can be modified, decorated, replaced or just
+     *         returned as a pass through
+     */
+    public Completable.CompletableOnSubscribe onCreate(Completable.CompletableOnSubscribe f) {
+        return f;
+    }
+
+    /**
+     * Invoked before {@link Completable#subscribe(Subscriber)} is about to be executed.
+     * <p>
+     * This can be used to decorate or replace the <code>onSubscribe</code> function or just perform extra
+     * logging, metrics and other such things and pass through the function.
+     *
+     * @param completableInstance the target completable instance
+     * @param onSubscribe
+     *            original {@link rx.Completable.CompletableOnSubscribe}<{@code T}> to be executed
+     * @return {@link rx.Completable.CompletableOnSubscribe}<{@code T}> function that can be modified, decorated, replaced or just
+     *         returned as a pass through
+     */
+    public Completable.CompletableOnSubscribe onSubscribeStart(Completable completableInstance, final Completable.CompletableOnSubscribe onSubscribe) {
+        // pass through by default
+        return onSubscribe;
+    }
+
+    /**
+     * Invoked after failed execution of {@link Completable#subscribe(Subscriber)} with thrown Throwable.
+     * <p>
+     * This is <em>not</em> errors emitted via {@link Subscriber#onError(Throwable)} but exceptions thrown when
+     * attempting to subscribe to a {@link Func1}<{@link Subscriber}{@code <T>}, {@link Subscription}>.
+     *
+     * @param e
+     *            Throwable thrown by {@link Completable#subscribe(Subscriber)}
+     * @return Throwable that can be decorated, replaced or just returned as a pass through
+     */
+    public Throwable onSubscribeError(Throwable e) {
+        // pass through by default
+        return e;
+    }
+
+    /**
+     * Invoked just as the operator functions is called to bind two operations together into a new
+     * {@link Completable} and the return value is used as the lifted function
+     * <p>
+     * This can be used to decorate or replace the {@link rx.Completable.CompletableOperator} instance or just perform extra
+     * logging, metrics and other such things and pass through the onSubscribe.
+     *
+     * @param lift
+     *            original {@link rx.Completable.CompletableOperator}{@code <R, T>}
+     * @return {@link rx.Completable.CompletableOperator}{@code <R, T>} function that can be modified, decorated, replaced or just
+     *         returned as a pass through
+     */
+    public Completable.CompletableOperator onLift(final Completable.CompletableOperator lift) {
+        return lift;
+    }
+}

src/main/java/rx/plugins/RxJavaCompletableExecutionHook.java

@@ -52,6 +52,7 @@ public class RxJavaPlugins {
     private final AtomicReference<RxJavaErrorHandler> errorHandler = new AtomicReference<RxJavaErrorHandler>();
     private final AtomicReference<RxJavaObservableExecutionHook> observableExecutionHook = new AtomicReference<RxJavaObservableExecutionHook>();
     private final AtomicReference<RxJavaSingleExecutionHook> singleExecutionHook = new AtomicReference<RxJavaSingleExecutionHook>();
+    private final AtomicReference<RxJavaCompletableExecutionHook> completableExecutionHook = new AtomicReference<RxJavaCompletableExecutionHook>();
     private final AtomicReference<RxJavaSchedulersHook> schedulersHook = new AtomicReference<RxJavaSchedulersHook>();
 
     /**
@@ -211,6 +212,52 @@ public void registerSingleExecutionHook(RxJavaSingleExecutionHook impl) {
         }
     }
 
+    /**
+     * Retrieves the instance of {@link RxJavaCompletableExecutionHook} to use based on order of precedence as
+     * defined in {@link RxJavaPlugins} class header.
+     * <p>
+     * Override the default by calling {@link #registerCompletableExecutionHook(RxJavaCompletableExecutionHook)}
+     * or by setting the property {@code rxjava.plugin.RxJavaCompletableExecutionHook.implementation} with the
+     * full classname to load.
+     *
+     * @return {@link RxJavaCompletableExecutionHook} implementation to use
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public RxJavaCompletableExecutionHook getCompletableExecutionHook() {
+        if (completableExecutionHook.get() == null) {
+            // check for an implementation from System.getProperty first
+            Object impl = getPluginImplementationViaProperty(RxJavaCompletableExecutionHook.class, System.getProperties());
+            if (impl == null) {
+                // nothing set via properties so initialize with default
+                completableExecutionHook.compareAndSet(null, new RxJavaCompletableExecutionHook() { });
+                // we don't return from here but call get() again in case of thread-race so the winner will always get returned
+            } else {
+                // we received an implementation from the system property so use it
+                completableExecutionHook.compareAndSet(null, (RxJavaCompletableExecutionHook) impl);
+            }
+        }
+        return completableExecutionHook.get();
+    }
+
+    /**
+     * Register an {@link RxJavaCompletableExecutionHook} implementation as a global override of any injected or
+     * default implementations.
+     *
+     * @param impl
+     *            {@link RxJavaCompletableExecutionHook} implementation
+     * @throws IllegalStateException
+     *             if called more than once or after the default was initialized (if usage occurs before trying
+     *             to register)
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public void registerCompletableExecutionHook(RxJavaCompletableExecutionHook impl) {
+        if (!completableExecutionHook.compareAndSet(null, impl)) {
+            throw new IllegalStateException("Another strategy was already registered: " + singleExecutionHook.get());
+        }
+    }
+
     /* test */ static Object getPluginImplementationViaProperty(Class<?> pluginClass, Properties propsIn) {
         // Make a defensive clone because traversal may fail with ConcurrentModificationException
         // if the properties get changed by something outside RxJava.