1.x: Added Single execution hooks

Author: phajduk

src/main/java/rx/Single.java

@@ -23,8 +23,8 @@
 import rx.internal.util.ScalarSynchronousSingle;
 import rx.internal.util.UtilityFunctions;
 import rx.observers.SafeSubscriber;
-import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
+import rx.plugins.RxJavaSingleExecutionHook;
 import rx.schedulers.Schedulers;
 import rx.singles.BlockingSingle;
 import rx.subscriptions.Subscriptions;
@@ -100,7 +100,7 @@ private Single(final Observable.OnSubscribe<T> f) {
         this.onSubscribe = f;
     }
 
-    static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
+    static final RxJavaSingleExecutionHook hook = RxJavaPlugins.getInstance().getSingleExecutionHook();
 
     /**
      * Returns a Single that will execute the specified function when a {@link SingleSubscriber} executes it or
@@ -129,7 +129,7 @@ private Single(final Observable.OnSubscribe<T> f) {
      * @see <a href="http://reactivex.io/documentation/operators/create.html">ReactiveX operators documentation: Create</a>
      */
     public static <T> Single<T> create(OnSubscribe<T> f) {
-        return new Single<T>(f); // TODO need hook 
+        return new Single<T>(hook.onCreate(f));
     }
 
     /**

src/main/java/rx/plugins/RxJavaPlugins.java

@@ -50,6 +50,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<RxJavaSchedulersHook> schedulersHook = new AtomicReference<RxJavaSchedulersHook>();
 
     /**
@@ -156,6 +157,48 @@ public void registerObservableExecutionHook(RxJavaObservableExecutionHook impl)
         }
     }
 
+    /**
+     * Retrieves the instance of {@link RxJavaSingleExecutionHook} to use based on order of precedence as
+     * defined in {@link RxJavaPlugins} class header.
+     * <p>
+     * Override the default by calling {@link #registerSingleExecutionHook(RxJavaSingleExecutionHook)}
+     * or by setting the property {@code rxjava.plugin.RxJavaSingleExecutionHook.implementation} with the
+     * full classname to load.
+     *
+     * @return {@link RxJavaSingleExecutionHook} implementation to use
+     */
+    public RxJavaSingleExecutionHook getSingleExecutionHook() {
+        if (singleExecutionHook.get() == null) {
+            // check for an implementation from System.getProperty first
+            Object impl = getPluginImplementationViaProperty(RxJavaSingleExecutionHook.class, System.getProperties());
+            if (impl == null) {
+                // nothing set via properties so initialize with default
+                singleExecutionHook.compareAndSet(null, RxJavaSingleExecutionHookDefault.getInstance());
+                // 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
+                singleExecutionHook.compareAndSet(null, (RxJavaSingleExecutionHook) impl);
+            }
+        }
+        return singleExecutionHook.get();
+    }
+
+    /**
+     * Register an {@link RxJavaSingleExecutionHook} implementation as a global override of any injected or
+     * default implementations.
+     *
+     * @param impl
+     *            {@link RxJavaSingleExecutionHook} implementation
+     * @throws IllegalStateException
+     *             if called more than once or after the default was initialized (if usage occurs before trying
+     *             to register)
+     */
+    public void registerSingleExecutionHook(RxJavaSingleExecutionHook impl) {
+        if (!singleExecutionHook.compareAndSet(null, impl)) {
+            throw new IllegalStateException("Another strategy was already registered: " + singleExecutionHook.get());
+        }
+    }
+
     /* test */ static Object getPluginImplementationViaProperty(Class<?> pluginClass, Properties props) {
         final String classSimpleName = pluginClass.getSimpleName();
         /*

src/main/java/rx/plugins/RxJavaSingleExecutionHook.java

@@ -0,0 +1,120 @@
+/**
+ * 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.Observable;
+import rx.Single;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.functions.Func1;
+
+/**
+ * Abstract ExecutionHook with invocations at different lifecycle points of {@link Single} 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 single so all behavior
+ * should be fast. If anything time-consuming is to be done it should be spawned asynchronously onto separate
+ * worker threads.
+ *
+ */
+public abstract class RxJavaSingleExecutionHook {
+    /**
+     * Invoked during the construction by {@link Single#create(Single.OnSubscribe)}
+     * <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-thru the function.
+     *
+     * @param f
+     *            original {@link Single.OnSubscribe}<{@code T}> to be executed
+     * @return {@link Single.OnSubscribe}<{@code T}> function that can be modified, decorated, replaced or just
+     *         returned as a pass-thru
+     */
+    public <T> Single.OnSubscribe<T> onCreate(Single.OnSubscribe<T> f) {
+        return f;
+    }
+
+    /**
+     * Invoked before {@link Single#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-thru the function.
+     *
+     * @param onSubscribe
+     *            original {@link Single.OnSubscribe}<{@code T}> to be executed
+     * @return {@link Single.OnSubscribe}<{@code T}> function that can be modified, decorated, replaced or just
+     *         returned as a pass-thru
+     */
+    public <T> Single.OnSubscribe<T> onSubscribeStart(Single<? extends T> singleInstance, final Single.OnSubscribe<T> onSubscribe) {
+        // pass-thru by default
+        return onSubscribe;
+    }
+
+    /**
+     * Invoked after successful execution of {@link Single#subscribe(Subscriber)} with returned
+     * {@link Subscription}.
+     * <p>
+     * This can be used to decorate or replace the {@link Subscription} instance or just perform extra logging,
+     * metrics and other such things and pass-thru the subscription.
+     *
+     * @param subscription
+     *            original {@link Subscription}
+     * @return {@link Subscription} subscription that can be modified, decorated, replaced or just returned as a
+     *         pass-thru
+     */
+    public <T> Subscription onSubscribeReturn(Subscription subscription) {
+        // pass-thru by default
+        return subscription;
+    }
+
+    /**
+     * Invoked after failed execution of {@link Single#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 Single#subscribe(Subscriber)}
+     * @return Throwable that can be decorated, replaced or just returned as a pass-thru
+     */
+    public <T> Throwable onSubscribeError(Throwable e) {
+        // pass-thru by default
+        return e;
+    }
+
+    /**
+     * Invoked just as the operator functions is called to bind two operations together into a new
+     * {@link Single} and the return value is used as the lifted function
+     * <p>
+     * This can be used to decorate or replace the {@link Observable.Operator} instance or just perform extra
+     * logging, metrics and other such things and pass-thru the onSubscribe.
+     *
+     * @param lift
+     *            original {@link Observable.Operator}{@code <R, T>}
+     * @return {@link Observable.Operator}{@code <R, T>} function that can be modified, decorated, replaced or just
+     *         returned as a pass-thru
+     */
+    public <T, R> Observable.Operator<? extends R, ? super T> onLift(final Observable.Operator<? extends R, ? super T> lift) {
+        return lift;
+    }
+}

src/main/java/rx/plugins/RxJavaSingleExecutionHookDefault.java

@@ -0,0 +1,28 @@
+/**
+ * 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;
+
+/**
+ * Default no-op implementation of {@link RxJavaSingleExecutionHook}
+ */
+/* package */class RxJavaSingleExecutionHookDefault extends RxJavaSingleExecutionHook {
+
+    private static RxJavaSingleExecutionHookDefault INSTANCE = new RxJavaSingleExecutionHookDefault();
+
+    public static RxJavaSingleExecutionHook getInstance() {
+        return INSTANCE;
+    }
+}

src/test/java/rx/plugins/RxJavaPluginsTest.java

@@ -107,6 +107,14 @@ public void testObservableExecutionHookViaRegisterMethod() {
         assertTrue(impl instanceof RxJavaObservableExecutionHookTestImpl);
     }
 
+    @Test
+    public void testSingleExecutionHookViaRegisterMethod() {
+        RxJavaPlugins p = new RxJavaPlugins();
+        p.registerSingleExecutionHook(new RxJavaSingleExecutionHookTestImpl());
+        RxJavaSingleExecutionHook impl = p.getSingleExecutionHook();
+        assertTrue(impl instanceof RxJavaSingleExecutionHookTestImpl);
+    }
+
     @Test
     public void testObservableExecutionHookViaProperty() {
         try {
@@ -238,6 +246,11 @@ public static class RxJavaObservableExecutionHookTestImpl extends RxJavaObservab
         // just use defaults
     }
 
+    // inside test so it is stripped from Javadocs
+    public static class RxJavaSingleExecutionHookTestImpl extends RxJavaSingleExecutionHook {
+        // just use defaults
+    }
+
     private static String getFullClassNameForTestClass(Class<?> cls) {
         return RxJavaPlugins.class.getPackage()
                                   .getName() + "." + RxJavaPluginsTest.class.getSimpleName() + "$" + cls.getSimpleName();