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

akarnokd · zsxwing · commit 4b61060d8d3d · 2016-06-01T10:06:52.000-07:00
Add option to safely subscribe a CompletableSubscriber / regular Subscriber and handle onXXX failures.
diff --git a/src/main/java/rx/Completable.java b/src/main/java/rx/Completable.java
diff --git a/src/main/java/rx/Single.java b/src/main/java/rx/Single.java
@@ -2024,7 +2024,7 @@ public void onSubscribe(Subscription d) {
                 serial.add(main);
                 child.add(serial);
 
-                other.subscribe(so);
+                other.unsafeSubscribe(so);
 
                 return main;
             }
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeConcat.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeConcat.java
@@ -130,7 +130,7 @@ void next() {
                 return;
             }
             
-            c.subscribe(inner);
+            c.unsafeSubscribe(inner);
         }
         
         final class ConcatInnerSubscriber implements CompletableSubscriber {
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeConcatArray.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeConcatArray.java
@@ -89,7 +89,7 @@ void next() {
                     return;
                 }
                 
-                a[idx].subscribe(this);
+                a[idx].unsafeSubscribe(this);
             } while (decrementAndGet() != 0);
         }
     }
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeConcatIterable.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeConcatIterable.java
@@ -128,7 +128,7 @@ void next() {
                     return;
                 }
                 
-                c.subscribe(this);
+                c.unsafeSubscribe(this);
             } while (decrementAndGet() != 0);
         }
     }
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeMerge.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeMerge.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
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeArray.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeArray.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);
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeDelayErrorArray.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeDelayErrorArray.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);
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeDelayErrorIterable.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeDelayErrorIterable.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);
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeIterable.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeMergeIterable.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);
diff --git a/src/main/java/rx/internal/operators/CompletableOnSubscribeTimeout.java b/src/main/java/rx/internal/operators/CompletableOnSubscribeTimeout.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) {
diff --git a/src/main/java/rx/observers/SafeCompletableSubscriber.java b/src/main/java/rx/observers/SafeCompletableSubscriber.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();
+    }
+}
diff --git a/src/main/java/rx/plugins/RxJavaCompletableExecutionHook.java b/src/main/java/rx/plugins/RxJavaCompletableExecutionHook.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 Wiki 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;
+    }
+}
diff --git a/src/main/java/rx/plugins/RxJavaPlugins.java b/src/main/java/rx/plugins/RxJavaPlugins.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.
diff --git a/src/test/java/rx/CompletableTest.java b/src/test/java/rx/CompletableTest.java
diff --git a/src/test/java/rx/SingleTest.java b/src/test/java/rx/SingleTest.java
diff --git a/src/test/java/rx/internal/operators/OnSubscribeCompletableTest.java b/src/test/java/rx/internal/operators/OnSubscribeCompletableTest.java
diff --git a/src/test/java/rx/plugins/RxJavaPluginsTest.java b/src/test/java/rx/plugins/RxJavaPluginsTest.java

Original file line number	Diff line number	Diff line change
`@@ -2024,7 +2024,7 @@ public void onSubscribe(Subscription d) {`
`2024`	`2024`	`serial.add(main);`
`2025`	`2025`	`child.add(serial);`
`2026`	`2026`
`2027`		`- other.subscribe(so);`
	`2027`	`+ other.unsafeSubscribe(so);`
`2028`	`2028`
`2029`	`2029`	`return main;`
`2030`	`2030`	`}`
Original file line number	Diff line number	Diff line change
`@@ -130,7 +130,7 @@ void next() {`
`130`	`130`	`return;`
`131`	`131`	`}`
`132`	`132`
`133`		`- c.subscribe(inner);`
	`133`	`+ c.unsafeSubscribe(inner);`
`134`	`134`	`}`
`135`	`135`
`136`	`136`	`final class ConcatInnerSubscriber implements CompletableSubscriber {`
Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,7 @@ void next() {`
`89`	`89`	`return;`
`90`	`90`	`}`
`91`	`91`
`92`		`- a[idx].subscribe(this);`
	`92`	`+ a[idx].unsafeSubscribe(this);`
`93`	`93`	`} while (decrementAndGet() != 0);`
`94`	`94`	`}`
`95`	`95`	`}`
Original file line number	Diff line number	Diff line change
`@@ -128,7 +128,7 @@ void next() {`
`128`	`128`	`return;`
`129`	`129`	`}`
`130`	`130`
`131`		`- c.subscribe(this);`
	`131`	`+ c.unsafeSubscribe(this);`
`132`	`132`	`} while (decrementAndGet() != 0);`
`133`	`133`	`}`
`134`	`134`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ public void call(final CompletableSubscriber s) {`
`54`	`54`	`}`
`55`	`55`	`}`
`56`	`56`
`57`		`- c.subscribe(new CompletableSubscriber() {`
	`57`	`+ c.unsafeSubscribe(new CompletableSubscriber() {`
`58`	`58`	`@Override`
`59`	`59`	`public void onSubscribe(Subscription d) {`
`60`	`60`	`set.add(d);`
Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public void call(final CompletableSubscriber s) {`
`51`	`51`	`continue;`
`52`	`52`	`}`
`53`	`53`
`54`		`- c.subscribe(new CompletableSubscriber() {`
	`54`	`+ c.unsafeSubscribe(new CompletableSubscriber() {`
`55`	`55`	`@Override`
`56`	`56`	`public void onSubscribe(Subscription d) {`
`57`	`57`	`set.add(d);`