Merge pull request #3696 from phajduk/SingleHooks

1.x: Added Single execution hooks

Author: akarnokd

src/main/java/rx/Single.java

@@ -29,6 +29,7 @@
 import rx.observers.SerializedSubscriber;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
+import rx.plugins.RxJavaSingleExecutionHook;
 import rx.schedulers.Schedulers;
 import rx.singles.BlockingSingle;
 import rx.subscriptions.Subscriptions;
@@ -101,7 +102,7 @@ private Single(final Observable.OnSubscribe<T> f) {
         this.onSubscribe = f;
     }
 
-    static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
+    static RxJavaSingleExecutionHook hook = RxJavaPlugins.getInstance().getSingleExecutionHook();
 
     /**
      * Returns a Single that will execute the specified function when a {@link SingleSubscriber} executes it or
@@ -130,7 +131,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));
     }
 
     /**
@@ -1607,14 +1608,12 @@ public final void onNext(T args) {
      * @param subscriber
      *            the Subscriber that will handle the emission or notification from the Single
      */
-    public final void unsafeSubscribe(Subscriber<? super T> subscriber) {
+    public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
         try {
             // new Subscriber so onStart it
             subscriber.onStart();
-            // TODO add back the hook
-            //            hook.onSubscribeStart(this, onSubscribe).call(subscriber);
-            onSubscribe.call(subscriber);
-            hook.onSubscribeReturn(subscriber);
+            hook.onSubscribeStart(this, onSubscribe).call(subscriber);
+            return hook.onSubscribeReturn(subscriber);
         } catch (Throwable e) {
             // special handling for certain Throwable/Error/Exception types
             Exceptions.throwIfFatal(e);
@@ -1631,6 +1630,7 @@ public final void unsafeSubscribe(Subscriber<? super T> subscriber) {
                 // TODO why aren't we throwing the hook's return value.
                 throw r;
             }
+            return Subscriptions.unsubscribed();
         }
     }
 
@@ -1722,9 +1722,7 @@ public final Subscription subscribe(Subscriber<? super T> subscriber) {
         // The code below is exactly the same an unsafeSubscribe but not used because it would add a sigificent depth to alreay huge call stacks.
         try {
             // allow the hook to intercept and/or decorate
-            // TODO add back the hook
-            //            hook.onSubscribeStart(this, onSubscribe).call(subscriber);
-            onSubscribe.call(subscriber);
+            hook.onSubscribeStart(this, onSubscribe).call(subscriber);
             return hook.onSubscribeReturn(subscriber);
         } catch (Throwable e) {
             // special handling for certain Throwable/Error/Exception types

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>();
 
     /**
@@ -68,6 +69,7 @@ public static RxJavaPlugins getInstance() {
     /* package accessible for unit tests */void reset() {
         INSTANCE.errorHandler.set(null);
         INSTANCE.observableExecutionHook.set(null);
+        INSTANCE.singleExecutionHook.set(null);
         INSTANCE.schedulersHook.set(null);
     }
 
@@ -156,6 +158,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 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 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 Observable.OnSubscribe}<{@code T}> to be executed
+     * @return {@link Observable.OnSubscribe}<{@code T}> function that can be modified, decorated, replaced or just
+     *         returned as a pass-thru
+     */
+    public <T> Observable.OnSubscribe<T> onSubscribeStart(Single<? extends T> singleInstance, final Observable.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}
+ */
+class RxJavaSingleExecutionHookDefault extends RxJavaSingleExecutionHook {
+
+    private static final RxJavaSingleExecutionHookDefault INSTANCE = new RxJavaSingleExecutionHookDefault();
+
+    public static RxJavaSingleExecutionHook getInstance() {
+        return INSTANCE;
+    }
+}